diff --git a/src/app/api/google-health/status/route.ts b/src/app/api/google-health/status/route.ts index 1fead2eda..76334bff5 100644 --- a/src/app/api/google-health/status/route.ts +++ b/src/app/api/google-health/status/route.ts @@ -3,7 +3,7 @@ import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; import { apiSuccess } from "@/lib/api-response"; import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load"; -import { googleHealthBackfillEndedList } from "@/lib/google-health/sync-core"; +import { googleHealthBackfillEndedList } from "@/lib/google-health/backfill-progress"; /** * Get Google Health connection status for the current user (v1.26.0). diff --git a/src/app/api/integrations/status/route.ts b/src/app/api/integrations/status/route.ts index 4bff2b04b..053f1dfad 100644 --- a/src/app/api/integrations/status/route.ts +++ b/src/app/api/integrations/status/route.ts @@ -55,7 +55,7 @@ import { loadConnectionHealth } from "@/lib/integrations/connection-health-load" import { googleHealthBackfillEndedList, type GoogleHealthBackfillEndedEntry, -} from "@/lib/google-health/sync-core"; +} from "@/lib/google-health/backfill-progress"; import type { ConnectionHealth } from "@/lib/integrations/connection-health"; export const dynamic = "force-dynamic"; diff --git a/src/lib/google-health/__tests__/backfill-bounds.test.ts b/src/lib/google-health/__tests__/backfill-bounds.test.ts index 581c46ff6..cc76b51ba 100644 --- a/src/lib/google-health/__tests__/backfill-bounds.test.ts +++ b/src/lib/google-health/__tests__/backfill-bounds.test.ts @@ -32,14 +32,16 @@ import { type GoogleHealthBackfillEnd, } from "../backfill-bounds"; import { - googleHealthBackfillEndedList, - googleHealthBackfillProgressAfterCompletion, - googleHealthBackfillRetryPending, - readGoogleHealthBackfillEnded, recordGoogleHealthBackfillEnd, runGoogleHealthCollection, runWithGoogleHealthBackfillResume, } from "../sync-core"; +import { + googleHealthBackfillEndedList, + googleHealthBackfillProgressAfterCompletion, + googleHealthBackfillRetryPending, + readGoogleHealthBackfillProgress, +} from "../backfill-progress"; import { forEachDataPointPage, GOOGLE_HEALTH_DATA_TYPES, @@ -312,7 +314,7 @@ describe("a collection ended short", () => { -1, )![0].data.backfillProgress as unknown; expect( - readGoogleHealthBackfillEnded(written).get("fetchHeartRate"), + readGoogleHealthBackfillProgress(written).ended.get("fetchHeartRate"), ).toEqual(note); // Waiting for another try is not where the history ends. expect(googleHealthBackfillEndedList(written)).toEqual([]); diff --git a/src/lib/google-health/__tests__/backfill-resume-progress.test.ts b/src/lib/google-health/__tests__/backfill-resume-progress.test.ts index 69e9715e9..80b2b2486 100644 --- a/src/lib/google-health/__tests__/backfill-resume-progress.test.ts +++ b/src/lib/google-health/__tests__/backfill-resume-progress.test.ts @@ -20,12 +20,14 @@ vi.mock("@/lib/logging/context", () => ({ })); import { - readGoogleHealthBackfillCursors, recordGoogleHealthBackfillPage, runGoogleHealthCollection, runWithGoogleHealthBackfillResume, - type GoogleHealthBackfillCursor, } from "../sync-core"; +import { + readGoogleHealthBackfillProgress, + type GoogleHealthBackfillCursor, +} from "../backfill-progress"; import { googleHealthSliceBudget, runWithGoogleHealthClientOutcome, @@ -124,7 +126,7 @@ describe("the resume edge of a page-by-page collection", () => { }); // And it reads back the way it was written. expect( - readGoogleHealthBackfillCursors(last.data.backfillProgress).get( + readGoogleHealthBackfillProgress(last.data.backfillProgress).cursors.get( "fetchHeartRate", ), ).toEqual(cursors.get("fetchHeartRate")); diff --git a/src/lib/google-health/__tests__/edge-filter-fallback.test.ts b/src/lib/google-health/__tests__/edge-filter-fallback.test.ts index 69a8cfc7e..1c827912a 100644 --- a/src/lib/google-health/__tests__/edge-filter-fallback.test.ts +++ b/src/lib/google-health/__tests__/edge-filter-fallback.test.ts @@ -46,13 +46,14 @@ import { GOOGLE_HEALTH_NO_PROGRESS_PAGES } from "../backfill-bounds"; import { googleHealthBackfillCursor, noteGoogleHealthEdgeFilterUnsupported, - readGoogleHealthBackfillCursors, - readGoogleHealthBackfillEdgeFilterUnsupported, recordGoogleHealthBackfillPage, recordGoogleHealthBackfillResumePoint, runWithGoogleHealthBackfillResume, - type GoogleHealthBackfillCursor, } from "../sync-core"; +import { + readGoogleHealthBackfillProgress, + type GoogleHealthBackfillCursor, +} from "../backfill-progress"; const MINUTE = 60_000; const NOW = Date.parse("2026-01-01T00:00:00Z"); @@ -566,7 +567,8 @@ describe("the note that a collection refuses the edge filter", () => { // Slice two: starts from what slice one left on disk. server.requests = []; - const unsupported = readGoogleHealthBackfillEdgeFilterUnsupported(progress); + const unsupported = + readGoogleHealthBackfillProgress(progress).edgeFilterUnsupported; await runWithGoogleHealthBackfillResume( "u1", new Set(), @@ -576,7 +578,7 @@ describe("the note that a collection refuses the edge filter", () => { const { rows } = await walk({ resumable: true, edge: cursor.edge }); expect(rows).toHaveLength(TOTAL - 19); }, - readGoogleHealthBackfillCursors(progress), + readGoogleHealthBackfillProgress(progress).cursors, unsupported, ); expect(server.requests.some((f) => f.includes("<="))).toBe(false); @@ -616,7 +618,7 @@ describe("the note that a collection refuses the edge filter", () => { page: 4, }); expect(first.advanced.has("fetchHeartRate")).toBe(true); - const saved = readGoogleHealthBackfillCursors(lastProgress()); + const saved = readGoogleHealthBackfillProgress(lastProgress()).cursors; expect(saved.get("fetchHeartRate")).toEqual({ ...cursor, resume: { token: "4", page: 4, reached: 4 }, @@ -628,8 +630,9 @@ describe("the note that a collection refuses the edge filter", () => { const back = await run(new Map(saved), { token: "2", page: 2 }); expect(back.advanced.size).toBe(0); expect( - readGoogleHealthBackfillCursors(lastProgress()).get("fetchHeartRate") - ?.resume, + readGoogleHealthBackfillProgress(lastProgress()).cursors.get( + "fetchHeartRate", + )?.resume, ).toEqual({ token: "2", page: 2, reached: 4 }); // A page written keeps the place; the next slice is handed it. @@ -644,7 +647,7 @@ describe("the note that a collection refuses the edge filter", () => { new Map(saved), new Set(["fetchHeartRate"]), ); - const next = readGoogleHealthBackfillCursors(lastProgress()); + const next = readGoogleHealthBackfillProgress(lastProgress()).cursors; expect(next.get("fetchHeartRate")?.resume).toEqual( saved.get("fetchHeartRate")?.resume, ); @@ -663,10 +666,12 @@ describe("the note that a collection refuses the edge filter", () => { it("reads only well-formed keys", () => { expect([ - ...readGoogleHealthBackfillEdgeFilterUnsupported({ + ...readGoogleHealthBackfillProgress({ edgeFilterUnsupported: ["fetchHeartRate", 3, "x".repeat(65)], - }), + }).edgeFilterUnsupported, ]).toEqual(["fetchHeartRate"]); - expect(readGoogleHealthBackfillEdgeFilterUnsupported(null).size).toBe(0); + expect( + readGoogleHealthBackfillProgress(null).edgeFilterUnsupported.size, + ).toBe(0); }); }); diff --git a/src/lib/google-health/backfill-progress.ts b/src/lib/google-health/backfill-progress.ts new file mode 100644 index 000000000..b4fb1caaf --- /dev/null +++ b/src/lib/google-health/backfill-progress.ts @@ -0,0 +1,279 @@ +/** + * The `backfillProgress` column of a Google Health connection: how far a + * full-history backfill has got, carried from one attempt to the next, and + * where a data type's history ended short of its beginning. + * + * One schema reads the column and one function writes it. The column is + * JSON the code itself wrote, but it outlives releases, so the read is + * lenient per entry: a malformed cursor or end is dropped on its own and the + * rest of the note still counts. A collection key longer than 64 characters + * is never one of ours and is dropped as well. + */ +import { z } from "zod"; + +import type { MeasurementType } from "@/generated/prisma/client"; +import { MeasurementType as MeasurementTypes } from "@/generated/prisma/enums"; +import { + googleHealthBackfillEndIsFinal, + googleHealthBackfillEndRetryAt, + type GoogleHealthBackfillEnd, +} from "./backfill-bounds"; + +/** + * How far a full-history backfill has got with one collection. + * + * Two grains. A collection walked to the end is noted in `done`, and the next + * attempt skips it. A collection read page by page (the spot readings: heart + * rate, weight, glucose, ...) also notes, after every page it writes, the + * edge it has reached: the oldest sample written so far when the walk runs + * newest first, as Google's list does. The next attempt asks Google only for + * the samples at or beyond that edge, so a collection that does not fit into + * one attempt is worked through over several instead of being read from its + * first page every time. + * + * The edge is a sample time, not Google's page token: page tokens are not + * documented to outlive the walk that received them, a sample time is a + * filter Google documents. The one sample on the edge itself is read twice + * and overwrites its own row. + * + * Before the edge existed only `done` was kept, and a heart-rate history of + * years that one attempt could not finish (its budget, a rate limit, one + * failed page among thousands) started again from the newest page on every + * attempt and never reached the end (#1220). + * + * The upper bound (`<=`) a newest-first walk resumes with is valid filter + * grammar but has not been seen working against the live API. A collection + * whose resumed request Google refuses with a 400 is noted in + * `edgeFilterUnsupported`, and its walks from then on read without the bound + * and pass over the samples beyond the edge themselves (see + * `forEachDataPointPage`). + */ +export interface GoogleHealthBackfillCursor { + /** The edge: the sample instant (ISO 8601) the walk has written up to. */ + at: string; + /** The order the walk returned samples in: newest first, or oldest first. */ + order: "desc" | "asc"; + /** + * Pages this collection has written across every attempt so far. Kept for + * the record only: it bounds nothing. It once capped the walk at ten + * thousand pages, which a dense heart-rate history passes in under three + * years, and since it never reset the backfill stopped there for good + * (#1220). The walk is bounded by its progress and the backfill horizon + * instead (`backfill-bounds.ts`). + */ + pages: number; + /** + * Only for a collection read without the edge bound: where its last walk + * stopped in the unbounded list (see `GoogleHealthResumePoint`), and the + * furthest position any walk of it has saved. A saved place counts as the + * slice's progress only when it is further than every one before it, so a + * chain of slices that keep passing over the same pages ends. + */ + resume?: { token: string; page: number; reached: number }; +} + +/** Everything the column holds, read. */ +export interface GoogleHealthBackfillProgress { + /** Collections walked to the end. */ + done: Set; + /** Per-collection edges. */ + cursors: Map; + /** Collections Google refused the edge filter on. */ + edgeFilterUnsupported: Set; + /** + * Collections whose backfill ended short of the start of their history + * (the horizon, or a walk that stopped getting anywhere), each with where + * it ended and why. Survives the completion stamp, so the integration card + * can say where history ends. + */ + ended: Map; +} + +const MAX_KEY_LENGTH = 64; + +/** An instant Date can read, normalised to ISO 8601. */ +const instant = z + .string() + .refine((s) => !Number.isNaN(Date.parse(s))) + .transform((s) => new Date(s).toISOString()); + +/** A whole, positive count; anything else reads as zero. */ +const count = z + .unknown() + .transform((v) => + typeof v === "number" && Number.isFinite(v) && v > 0 ? Math.trunc(v) : 0, + ); + +const resumeSchema = z + .object({ token: z.string().min(1).max(4096), page: count, reached: count }) + .transform((r) => + r.page > 0 + ? { token: r.token, page: r.page, reached: Math.max(r.reached, r.page) } + : undefined, + ); + +const cursorSchema = z + .object({ + at: instant, + order: z.unknown().transform((v) => (v === "asc" ? "asc" : "desc")), + pages: count, + resume: resumeSchema.optional().catch(undefined), + }) + .transform(({ resume, ...c }): GoogleHealthBackfillCursor => + resume ? { ...c, resume } : c, + ); + +const endSchema = z + .object({ + type: z.enum(MeasurementTypes), + at: instant, + reason: z.enum(["horizon", "no_progress"]), + attempts: z.number().int().min(1).optional().catch(undefined), + endedAt: instant.optional().catch(undefined), + }) + .transform(({ attempts, endedAt, ...e }): GoogleHealthBackfillEnd => ({ + ...e, + type: e.type as MeasurementType, + ...(attempts !== undefined ? { attempts } : {}), + ...(endedAt !== undefined ? { endedAt } : {}), + })); + +const record = z.record(z.string(), z.unknown()).catch({}); + +const keyList = z + .array(z.unknown()) + .catch([]) + .transform( + (keys) => + new Set( + keys.filter( + (key): key is string => + typeof key === "string" && key.length <= MAX_KEY_LENGTH, + ), + ), + ); + +/** Entries of a keyed record that parse, by their own schema, one by one. */ +function keyedEntries(value: unknown, schema: z.ZodType): Map { + const out = new Map(); + for (const [key, raw] of Object.entries(record.parse(value))) { + if (key.length > MAX_KEY_LENGTH) continue; + const parsed = schema.safeParse(raw); + if (parsed.success) out.set(key, parsed.data); + } + return out; +} + +/** Read the `backfillProgress` column. */ +export function readGoogleHealthBackfillProgress( + value: unknown, +): GoogleHealthBackfillProgress { + const column = record.parse(value); + return { + done: keyList.parse(column.done), + cursors: keyedEntries(column.cursors, cursorSchema), + edgeFilterUnsupported: keyList.parse(column.edgeFilterUnsupported), + ended: keyedEntries(column.ended, endSchema), + }; +} + +/** The ended collections as the column stores them. */ +function endedJson(ended: Map) { + return Object.fromEntries( + [...ended].map(([key, e]) => [ + key, + { + type: e.type, + at: e.at, + reason: e.reason, + ...(e.attempts !== undefined ? { attempts: e.attempts } : {}), + ...(e.endedAt !== undefined ? { endedAt: e.endedAt } : {}), + }, + ]), + ); +} + +/** + * The column's value for a backfill in progress. The two optional parts are + * written only once there is something to say, so a backfill that never met + * them keeps the shape it always had. + */ +export function googleHealthBackfillProgressJson( + progress: GoogleHealthBackfillProgress, +) { + const { done, cursors, edgeFilterUnsupported, ended } = progress; + return { + done: [...done], + cursors: Object.fromEntries( + [...cursors].map(([key, c]) => [ + key, + { + at: c.at, + order: c.order, + pages: c.pages, + ...(c.resume ? { resume: c.resume } : {}), + }, + ]), + ), + ...(edgeFilterUnsupported.size > 0 + ? { edgeFilterUnsupported: [...edgeFilterUnsupported] } + : {}), + ...(ended.size > 0 ? { ended: endedJson(ended) } : {}), + }; +} + +/** + * The column's value to keep once the backfill is stamped complete: only + * where collections ended short for good, or nothing at all. + */ +export function googleHealthBackfillProgressAfterCompletion( + ended: Map, +): { ended: ReturnType } | null { + const final = new Map( + [...ended].filter(([, e]) => googleHealthBackfillEndIsFinal(e)), + ); + return final.size > 0 ? { ended: endedJson(final) } : null; +} + +/** + * Where each data type's imported history ends short of its beginning, and + * why, as the status routes publish it (`backfillEnded`), by type. + */ +export interface GoogleHealthBackfillEndedEntry { + type: MeasurementType; + /** The oldest instant the imported history reaches back to. */ + endsAt: string; + reason: GoogleHealthBackfillEnd["reason"]; +} + +/** + * The column's ended collections, as published: only the final ends. One + * still waiting for another attempt is not where its history ends yet. + */ +export function googleHealthBackfillEndedList( + value: unknown, +): GoogleHealthBackfillEndedEntry[] { + return [...readGoogleHealthBackfillProgress(value).ended.values()] + .filter(googleHealthBackfillEndIsFinal) + .map((e) => ({ type: e.type, endsAt: e.at, reason: e.reason })) + .sort((a, b) => a.type.localeCompare(b.type)); +} + +/** + * The collections waiting for another attempt after one ended without + * progress, and the earliest instant (epoch ms) one of them is due, or null + * when none waits. + */ +export function googleHealthBackfillRetryPending( + ended: Map, +): { keys: string[]; dueAt: number } | null { + const keys: string[] = []; + let dueAt = Infinity; + for (const [key, e] of ended) { + const at = googleHealthBackfillEndRetryAt(e); + if (at === null) continue; + keys.push(key); + dueAt = Math.min(dueAt, at); + } + return keys.length > 0 ? { keys, dueAt } : null; +} diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index b0ee51d1f..25cb8b885 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -18,7 +18,6 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { prisma } from "@/lib/db"; import type { MeasurementType } from "@/generated/prisma/client"; -import { MeasurementType as MeasurementTypes } from "@/generated/prisma/enums"; import { encrypt, decrypt } from "@/lib/crypto"; import { annotate, getEvent } from "@/lib/logging/context"; import type { GoogleHealthCollectionKey } from "./collection-keys"; @@ -27,6 +26,10 @@ import { googleHealthBackfillEndRetryAt, type GoogleHealthBackfillEnd, } from "./backfill-bounds"; +import { + googleHealthBackfillProgressJson, + type GoogleHealthBackfillCursor, +} from "./backfill-progress"; import { userDayKey } from "@/lib/tz/format"; import { isP2002 } from "@/lib/prisma-errors"; import { @@ -411,60 +414,6 @@ export async function runWithGoogleHealthSyncCycle( }; } -/** - * How far a full-history backfill has got, carried from one attempt of the - * backfill to the next. - * - * Two grains. A collection walked to the end is noted in `done`, and the next - * attempt skips it. A collection read page by page (the spot readings: heart - * rate, weight, glucose, ...) also notes, after every page it writes, the - * edge it has reached: the oldest sample written so far when the walk runs - * newest first, as Google's list does. The next attempt asks Google only for - * the samples at or beyond that edge, so a collection that does not fit into - * one attempt is worked through over several instead of being read from its - * first page every time. - * - * The edge is a sample time, not Google's page token: page tokens are not - * documented to outlive the walk that received them, a sample time is a - * filter Google documents. The one sample on the edge itself is read twice - * and overwrites its own row. - * - * Before the edge existed only `done` was kept, and a heart-rate history of - * years that one attempt could not finish (its budget, a rate limit, one - * failed page among thousands) started again from the newest page on every - * attempt and never reached the end (#1220). - * - * The upper bound (`<=`) a newest-first walk resumes with is valid filter - * grammar but has not been seen working against the live API. A collection - * whose resumed request Google refuses with a 400 is noted in - * `edgeFilterUnsupported`, and its walks from then on read without the bound - * and pass over the samples beyond the edge themselves (see - * `forEachDataPointPage`). - */ -export interface GoogleHealthBackfillCursor { - /** The edge: the sample instant (ISO 8601) the walk has written up to. */ - at: string; - /** The order the walk returned samples in: newest first, or oldest first. */ - order: "desc" | "asc"; - /** - * Pages this collection has written across every attempt so far. Kept for - * the record only: it bounds nothing. It once capped the walk at ten - * thousand pages, which a dense heart-rate history passes in under three - * years, and since it never reset the backfill stopped there for good - * (#1220). The walk is bounded by its progress and the backfill horizon - * instead (`backfill-bounds.ts`). - */ - pages: number; - /** - * Only for a collection read without the edge bound: where its last walk - * stopped in the unbounded list (see `GoogleHealthResumePoint`), and the - * furthest position any walk of it has saved. A saved place counts as the - * slice's progress only when it is further than every one before it, so a - * chain of slices that keep passing over the same pages ends. - */ - resume?: { token: string; page: number; reached: number }; -} - /** What one collection came to in one backfill attempt. */ export interface GoogleHealthBackfillCollectionReport { key: string; @@ -504,203 +453,6 @@ interface BackfillResumeTracker { } const backfillResumeStorage = new AsyncLocalStorage(); -function progressRecord(value: unknown): Record { - return value && typeof value === "object" && !Array.isArray(value) - ? (value as Record) - : {}; -} - -/** Read the `backfillProgress` column into the set of finished collections. */ -export function readGoogleHealthBackfillDone(value: unknown): Set { - const done = progressRecord(value).done; - return new Set( - Array.isArray(done) - ? done.filter( - (key): key is string => typeof key === "string" && key.length <= 64, - ) - : [], - ); -} - -/** Read the `backfillProgress` column into the per-collection edges. */ -export function readGoogleHealthBackfillCursors( - value: unknown, -): Map { - const out = new Map(); - const cursors = progressRecord(progressRecord(value).cursors); - for (const [key, raw] of Object.entries(cursors)) { - if (key.length > 64) continue; - const c = progressRecord(raw); - const at = typeof c.at === "string" ? c.at : null; - if (!at || Number.isNaN(Date.parse(at))) continue; - const r = progressRecord(c.resume); - const count = (v: unknown): number => - typeof v === "number" && Number.isFinite(v) && v > 0 ? Math.trunc(v) : 0; - const resume = - typeof r.token === "string" && - r.token.length > 0 && - r.token.length <= 4096 && - count(r.page) > 0 - ? { - token: r.token, - page: count(r.page), - reached: Math.max(count(r.reached), count(r.page)), - } - : undefined; - out.set(key, { - at: new Date(at).toISOString(), - order: c.order === "asc" ? "asc" : "desc", - pages: count(c.pages), - ...(resume ? { resume } : {}), - }); - } - return out; -} - -/** - * Read the `backfillProgress` column into the collections whose backfill - * ended short of the start of their history (the horizon, or a walk that - * stopped getting anywhere), each with where it ended and why. Survives the - * completion stamp, so the integration card can say where history ends. - */ -export function readGoogleHealthBackfillEnded( - value: unknown, -): Map { - const out = new Map(); - const ended = progressRecord(progressRecord(value).ended); - for (const [key, raw] of Object.entries(ended)) { - if (key.length > 64) continue; - const e = progressRecord(raw); - const at = typeof e.at === "string" ? e.at : null; - if (!at || Number.isNaN(Date.parse(at))) continue; - if (e.reason !== "horizon" && e.reason !== "no_progress") continue; - if (typeof e.type !== "string" || !(e.type in MeasurementTypes)) { - continue; - } - const attempts = - typeof e.attempts === "number" && - Number.isInteger(e.attempts) && - e.attempts >= 1 - ? e.attempts - : undefined; - const endedAt = - typeof e.endedAt === "string" && !Number.isNaN(Date.parse(e.endedAt)) - ? new Date(e.endedAt).toISOString() - : undefined; - out.set(key, { - type: e.type as MeasurementType, - at: new Date(at).toISOString(), - reason: e.reason, - ...(attempts !== undefined ? { attempts } : {}), - ...(endedAt !== undefined ? { endedAt } : {}), - }); - } - return out; -} - -/** - * Where each data type's imported history ends short of its beginning, and - * why, as the status routes publish it (`backfillEnded`), by type. - */ -export interface GoogleHealthBackfillEndedEntry { - type: MeasurementType; - /** The oldest instant the imported history reaches back to. */ - endsAt: string; - reason: GoogleHealthBackfillEnd["reason"]; -} - -/** - * The `backfillProgress` column's ended collections, as published: only the - * final ends. One still waiting for another attempt is not where its history - * ends yet. - */ -export function googleHealthBackfillEndedList( - value: unknown, -): GoogleHealthBackfillEndedEntry[] { - return [...readGoogleHealthBackfillEnded(value).values()] - .filter(googleHealthBackfillEndIsFinal) - .map((e) => ({ type: e.type, endsAt: e.at, reason: e.reason })) - .sort((a, b) => a.type.localeCompare(b.type)); -} - -/** - * The `backfillProgress` value to keep once the backfill is stamped complete: - * only where collections ended short, or nothing at all. - */ -export function googleHealthBackfillProgressAfterCompletion( - ended: Map, -): { - ended: Record; -} | null { - const final = new Map( - [...ended].filter(([, e]) => googleHealthBackfillEndIsFinal(e)), - ); - return final.size > 0 ? { ended: endedJson(final) } : null; -} - -/** One ended collection as the column stores it. */ -type GoogleHealthBackfillEndJson = { - type: string; - at: string; - reason: string; - attempts?: number; - endedAt?: string; -}; - -/** The ended collections as the column stores them. */ -function endedJson( - ended: Map, -): Record { - return Object.fromEntries( - [...ended].map(([key, e]) => [ - key, - { - type: e.type, - at: e.at, - reason: e.reason, - ...(e.attempts !== undefined ? { attempts: e.attempts } : {}), - ...(e.endedAt !== undefined ? { endedAt: e.endedAt } : {}), - }, - ]), - ); -} - -/** - * The collections waiting for another attempt after one ended without - * progress, and the earliest instant (epoch ms) one of them is due, or null - * when none waits. - */ -export function googleHealthBackfillRetryPending( - ended: Map, -): { keys: string[]; dueAt: number } | null { - const keys: string[] = []; - let dueAt = Infinity; - for (const [key, e] of ended) { - const at = googleHealthBackfillEndRetryAt(e); - if (at === null) continue; - keys.push(key); - dueAt = Math.min(dueAt, at); - } - return keys.length > 0 ? { keys, dueAt } : null; -} - -/** - * Read the `backfillProgress` column into the collections Google refused the - * edge filter on. - */ -export function readGoogleHealthBackfillEdgeFilterUnsupported( - value: unknown, -): Set { - const keys = progressRecord(value).edgeFilterUnsupported; - return new Set( - Array.isArray(keys) - ? keys.filter( - (key): key is string => typeof key === "string" && key.length <= 64, - ) - : [], - ); -} - /** * Write the tracker's progress with one change applied, and apply it to the * tracker only once the write has landed: progress that is not on disk is @@ -721,26 +473,12 @@ async function persistBackfillProgress( await prisma.googleHealthConnection.update({ where: { userId: tracker.userId }, data: { - backfillProgress: { - done: [...next.done], - cursors: Object.fromEntries( - [...next.cursors].map(([key, c]) => [ - key, - { - at: c.at, - order: c.order, - pages: c.pages, - ...(c.resume ? { resume: c.resume } : {}), - }, - ]), - ), - // Only once there is something to say, so a backfill that never met - // the refusal keeps the shape it always had. - ...(unsupported.size > 0 - ? { edgeFilterUnsupported: [...unsupported] } - : {}), - ...(ended.size > 0 ? { ended: endedJson(ended) } : {}), - }, + backfillProgress: googleHealthBackfillProgressJson({ + done: next.done, + cursors: next.cursors, + edgeFilterUnsupported: unsupported, + ended, + }), }, }); if (unsupported !== tracker.edgeFilterUnsupported) { diff --git a/src/lib/jobs/__tests__/google-health-backfill.test.ts b/src/lib/jobs/__tests__/google-health-backfill.test.ts index d7c802d29..f89980edc 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -63,12 +63,13 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => { return { ...actual, safeFetch: safeFetchMock }; }); -vi.mock("@/lib/google-health/sync-core", () => ({ - GOOGLE_HEALTH_INTEGRATION_KEY: "google-health", - readGoogleHealthBackfillDone: () => new Set(resumeState.doneBefore), - readGoogleHealthBackfillCursors: () => new Map(), - readGoogleHealthBackfillEdgeFilterUnsupported: () => new Set(), - readGoogleHealthBackfillEnded: () => new Map(), +vi.mock("@/lib/google-health/backfill-progress", () => ({ + readGoogleHealthBackfillProgress: () => ({ + done: new Set(resumeState.doneBefore), + cursors: new Map(), + edgeFilterUnsupported: new Set(), + ended: new Map(), + }), googleHealthBackfillProgressAfterCompletion: (ended: Map) => ended.size > 0 ? { ended: Object.fromEntries(ended) } : null, googleHealthBackfillRetryPending: ( @@ -79,6 +80,10 @@ vi.mock("@/lib/google-health/sync-core", () => ({ .map(([key]) => key); return keys.length > 0 ? { keys, dueAt: resumeState.retryDueAt } : null; }, +})); + +vi.mock("@/lib/google-health/sync-core", () => ({ + GOOGLE_HEALTH_INTEGRATION_KEY: "google-health", runWithGoogleHealthBackfillResume: async ( _userId: string, done: Set, diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index b5e94201f..a23a6b016 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -20,15 +20,12 @@ import { INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, integrationBackfillSourceOptions, } from "@/lib/jobs/integration-backfill-admission"; +import { runWithGoogleHealthBackfillResume } from "@/lib/google-health/sync-core"; import { googleHealthBackfillProgressAfterCompletion, - readGoogleHealthBackfillCursors, - readGoogleHealthBackfillDone, - readGoogleHealthBackfillEdgeFilterUnsupported, - readGoogleHealthBackfillEnded, googleHealthBackfillRetryPending, - runWithGoogleHealthBackfillResume, -} from "@/lib/google-health/sync-core"; + readGoogleHealthBackfillProgress, +} from "@/lib/google-health/backfill-progress"; import { googleHealthBackfillEndIsFinal } from "@/lib/google-health/backfill-bounds"; import { isGoogleHealthParked, @@ -198,14 +195,12 @@ export async function runGoogleHealthBackfillForUser( }); return { imported: 0 }; } - const done = readGoogleHealthBackfillDone(connection?.backfillProgress); - const cursors = readGoogleHealthBackfillCursors(connection?.backfillProgress); - const edgeFilterUnsupported = readGoogleHealthBackfillEdgeFilterUnsupported( - connection?.backfillProgress, - ); - const endedBefore = readGoogleHealthBackfillEnded( - connection?.backfillProgress, - ); + const { + done, + cursors, + edgeFilterUnsupported, + ended: endedBefore, + } = readGoogleHealthBackfillProgress(connection?.backfillProgress); // A collection that ended short for good has read all it is going to, so // it counts as finished alongside the ones read to the end. One parked // after a first end without progress does not: it waits for another try. diff --git a/tests/integration/google-health-backfill-progress-bound.test.ts b/tests/integration/google-health-backfill-progress-bound.test.ts index df945ae9f..1d7b5b604 100644 --- a/tests/integration/google-health-backfill-progress-bound.test.ts +++ b/tests/integration/google-health-backfill-progress-bound.test.ts @@ -337,7 +337,7 @@ describe("Google Health backfill bounded by progress, not pages (#1220)", () => const { GOOGLE_HEALTH_NO_PROGRESS_PAGES } = await import("@/lib/google-health/backfill-bounds"); const { googleHealthBackfillEndedList } = - await import("@/lib/google-health/sync-core"); + await import("@/lib/google-health/backfill-progress"); const walk = 1 + GOOGLE_HEALTH_NO_PROGRESS_PAGES; // First attempt: the first page and the twenty that repeat it. Not @@ -413,7 +413,7 @@ describe("Google Health backfill bounded by progress, not pages (#1220)", () => await runChain(5); const { googleHealthBackfillEndedList } = - await import("@/lib/google-health/sync-core"); + await import("@/lib/google-health/backfill-progress"); const c = await connection(); expect(c.backfillCompletedAt).toBeInstanceOf(Date); const ended = googleHealthBackfillEndedList(c.backfillProgress);