refactor(google-health): read and write backfill progress through one schema

The backfillProgress column was read by four hand-rolled walkers over
untyped records and written inline in the sync core. One module now owns
the column: a Zod schema reads it, lenient per entry so a malformed cursor
or end is dropped on its own, and one function writes it. The sync core
keeps only the bookkeeping of a running backfill.
This commit is contained in:
Marc-André Bombeck committed 2026-10-20 03:34:30 +02:00
1 parent 435099dc4f
commit bfbd88b30b
10 files changed
+342 -316

No files matched your search

+1 -1
View File
@@ -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).
+1 -1
View File
@@ -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";
@@ -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([]);
@@ -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"));
@@ -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);
});
});
+279
View File
@@ -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<string>;
/** Per-collection edges. */
cursors: Map<string, GoogleHealthBackfillCursor>;
/** Collections Google refused the edge filter on. */
edgeFilterUnsupported: Set<string>;
/**
* 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<string, GoogleHealthBackfillEnd>;
}
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<T>(value: unknown, schema: z.ZodType<T>): Map<string, T> {
const out = new Map<string, T>();
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<string, GoogleHealthBackfillEnd>) {
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<string, GoogleHealthBackfillEnd>,
): { ended: ReturnType<typeof endedJson> } | 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<string, GoogleHealthBackfillEnd>,
): { 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;
}
+10 -272
View File
@@ -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<T>(
};
}
/**
* 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<BackfillResumeTracker>();
function progressRecord(value: unknown): Record<string, unknown> {
return value && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: {};
}
/** Read the `backfillProgress` column into the set of finished collections. */
export function readGoogleHealthBackfillDone(value: unknown): Set<string> {
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<string, GoogleHealthBackfillCursor> {
const out = new Map<string, GoogleHealthBackfillCursor>();
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<string, GoogleHealthBackfillEnd> {
const out = new Map<string, GoogleHealthBackfillEnd>();
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<string, GoogleHealthBackfillEnd>,
): {
ended: Record<string, GoogleHealthBackfillEndJson>;
} | 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<string, GoogleHealthBackfillEnd>,
): Record<string, GoogleHealthBackfillEndJson> {
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<string, GoogleHealthBackfillEnd>,
): { 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<string> {
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) {
@@ -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<string>(resumeState.doneBefore),
readGoogleHealthBackfillCursors: () => new Map(),
readGoogleHealthBackfillEdgeFilterUnsupported: () => new Set<string>(),
readGoogleHealthBackfillEnded: () => new Map(),
vi.mock("@/lib/google-health/backfill-progress", () => ({
readGoogleHealthBackfillProgress: () => ({
done: new Set<string>(resumeState.doneBefore),
cursors: new Map(),
edgeFilterUnsupported: new Set<string>(),
ended: new Map(),
}),
googleHealthBackfillProgressAfterCompletion: (ended: Map<string, unknown>) =>
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<string>,
+9 -14
View File
@@ -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.
@@ -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);