refactor(sync): move the delta feed's domain pages out of its route

This commit is contained in:
Marc-André Bombeck committed 2026-10-23 14:45:26 +02:00
1 parent 1b5e586a06
commit ad27e5b89d
8 files changed
+722 -568

No files matched your search

+229
View File
@@ -0,0 +1,229 @@
/**
* Turn each domain's rows into its wire block: tombstones for the
* soft-deleted rows, upserts for the live ones.
*/
import { prisma } from "@/lib/db";
import {
findCompactionTombstones,
isFoldedWindowCandidate,
loadFoldTimezone,
} from "@/lib/measurements/folded-window";
import { readNote } from "@/lib/crypto/note-cipher";
import { levelAForWire } from "@/lib/mood/level-a";
import { contextForWire } from "@/lib/mood/context";
import {
toCycleDayLogDTO,
toMenstrualCycleDTO,
type CycleDayLogDTO,
type MenstrualCycleDTO,
} from "@/lib/cycle/dto";
import type {
CycleDayRow,
CycleRow,
IntakeRow,
MeasurementRow,
MoodRow,
} from "./read-rows";
import type {
CycleDayTombstone,
DomainBlock,
IdTombstone,
IntakeUpsert,
MeasurementTombstone,
MeasurementUpsert,
MoodUpsert,
} from "./wire";
/** A domain's page: the rows inside the limit, and whether more follow. */
export function pageOf<R>(
rows: R[],
limit: number,
): { page: R[]; hasMore: boolean } {
const hasMore = rows.length > limit;
return { page: hasMore ? rows.slice(0, limit) : rows, hasMore };
}
function idTombstone(row: {
id: string;
syncVersion: number;
deletedAt: Date;
updatedAt: Date;
}): IdTombstone {
return {
id: row.id,
syncVersion: row.syncVersion,
deletedAt: row.deletedAt.toISOString(),
updatedAt: row.updatedAt.toISOString(),
};
}
/**
* Compaction tombstones are not deletions anyone made. The folds used to
* soft-delete the raw Apple Health samples they folded into an hourly or
* daily `stats:` row, and those rows rode this feed as tombstones for 75
* days. They are left out, by the same classifier the backlog purge uses
* (`folded-window.ts`); the cursor still advances past them, so the keyset
* walk is unchanged. A person's own deletions are reported exactly as
* before.
*/
async function compactionIds(
userId: string,
page: MeasurementRow[],
): Promise<Set<string>> {
const compaction = new Set<string>();
const tombstonesOnPage = page.filter((row) => row.deletedAt !== null);
if (tombstonesOnPage.length === 0) return compaction;
const candidates = tombstonesOnPage.map((row) => ({
type: row.type,
source: row.source,
externalId: row.externalId,
measuredAt: row.measuredAt,
deletedAt: row.deletedAt,
}));
if (candidates.some((row) => isFoldedWindowCandidate(row, row.deletedAt!))) {
const classA = await findCompactionTombstones(
prisma,
userId,
await loadFoldTimezone(prisma, userId),
candidates,
);
for (const index of classA) compaction.add(tombstonesOnPage[index].id);
}
return compaction;
}
export async function measurementBlock(
userId: string,
page: MeasurementRow[],
): Promise<DomainBlock<MeasurementUpsert, MeasurementTombstone>> {
const compaction = await compactionIds(userId, page);
const upserts: MeasurementUpsert[] = [];
const tombstones: MeasurementTombstone[] = [];
for (const row of page) {
if (compaction.has(row.id)) continue;
if (row.deletedAt) {
tombstones.push({
id: row.id,
externalId: row.externalId,
syncVersion: row.syncVersion,
deletedAt: row.deletedAt.toISOString(),
updatedAt: row.updatedAt.toISOString(),
});
} else {
// Every row carries its `source`. RECOVERY_SCORE is written by BOTH the
// WHOOP-native sync (`source = WHOOP`) and the COMPUTED proxy job
// (`source = COMPUTED`) for the same day; the client must resolve to ONE
// canonical recovery series (WHOOP wins when present, COMPUTED is the
// fallback) and never chart the proxy and the native value as two
// competing series. The feed faithfully mirrors both rows (a delta-sync
// mirror cannot drop a row mid-keyset without a tombstone); the
// canonical pick is the same WHOOP-over-COMPUTED rule the server's
// wellness tile + doctor PDF apply (`resolveCanonicalRecovery`).
upserts.push({
id: row.id,
externalId: row.externalId,
type: row.type,
value: row.value,
unit: row.unit,
measuredAt: row.measuredAt.toISOString(),
source: row.source,
notes: readNote(row.notesEncrypted, row.notes),
syncVersion: row.syncVersion,
updatedAt: row.updatedAt.toISOString(),
});
}
}
return { upserts, tombstones };
}
export function moodBlock(
page: MoodRow[],
): DomainBlock<MoodUpsert, IdTombstone> {
const upserts: MoodUpsert[] = [];
const tombstones: IdTombstone[] = [];
for (const row of page) {
if (row.deletedAt) {
tombstones.push(idTombstone({ ...row, deletedAt: row.deletedAt }));
} else {
upserts.push({
id: row.id,
date: row.date,
mood: row.mood,
score: row.score,
// Under the same wire keys the write path accepts, so a client reads
// back exactly what it can post.
...levelAForWire(row),
context: row.context ? contextForWire(row.context) : null,
tags: row.tags,
note: readNote(row.noteEncrypted, row.note),
moodLoggedAt: row.moodLoggedAt.toISOString(),
source: row.source,
syncVersion: row.syncVersion,
updatedAt: row.updatedAt.toISOString(),
});
}
}
return { upserts, tombstones };
}
export function intakeBlock(
page: IntakeRow[],
): DomainBlock<IntakeUpsert, IdTombstone> {
const upserts: IntakeUpsert[] = [];
const tombstones: IdTombstone[] = [];
for (const row of page) {
if (row.deletedAt) {
tombstones.push(idTombstone({ ...row, deletedAt: row.deletedAt }));
} else {
upserts.push({
id: row.id,
medicationId: row.medicationId,
scheduledFor: row.scheduledFor.toISOString(),
takenAt: row.takenAt?.toISOString() ?? null,
skipped: row.skipped,
source: row.source,
autoRecorded: row.autoRecorded,
syncVersion: row.syncVersion,
updatedAt: row.updatedAt.toISOString(),
});
}
}
return { upserts, tombstones };
}
export function cycleDayBlock(
page: CycleDayRow[],
): DomainBlock<CycleDayLogDTO, CycleDayTombstone> {
const upserts: CycleDayLogDTO[] = [];
const tombstones: CycleDayTombstone[] = [];
for (const row of page) {
if (row.deletedAt) {
tombstones.push({
id: row.id,
externalId: row.externalId,
syncVersion: row.syncVersion,
deletedAt: row.deletedAt.toISOString(),
updatedAt: row.updatedAt.toISOString(),
});
} else {
upserts.push(toCycleDayLogDTO(row));
}
}
return { upserts, tombstones };
}
export function cycleBlock(
page: CycleRow[],
): DomainBlock<MenstrualCycleDTO, IdTombstone> {
const upserts: MenstrualCycleDTO[] = [];
const tombstones: IdTombstone[] = [];
for (const row of page) {
if (row.deletedAt) {
tombstones.push(idTombstone({ ...row, deletedAt: row.deletedAt }));
} else {
upserts.push(toMenstrualCycleDTO(row));
}
}
return { upserts, tombstones };
}
+218
View File
@@ -0,0 +1,218 @@
/**
* One pull of the multi-domain delta feed behind `GET /api/sync/changes`.
*
* The route authenticates, meters and parses the query; this module decides
* whether the cursor can still be caught up, reads one page of every domain
* past its watermark, and answers the blocks and the next cursor. The feed
* contract is described on the route.
*/
import { prisma } from "@/lib/db";
import { annotate } from "@/lib/logging/context";
import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client";
import {
decodeCursor,
decodeCursorIssuedAt,
encodeCursor,
SYNC_DOMAINS,
type SyncCursor,
type SyncDomain,
} from "@/lib/sync/cursor";
import { isModuleEnabled } from "@/lib/modules/gate";
import { readDomainRows } from "./read-rows";
import {
cycleBlock,
cycleDayBlock,
intakeBlock,
measurementBlock,
moodBlock,
pageOf,
} from "./domain-pages";
import type { SyncChangesBody } from "./wire";
const DAY_MS = 86_400_000;
/**
* Advance a domain's watermark to the last row of its page. A domain that
* returned no rows keeps its prior watermark untouched so a future pull does
* not re-scan from the start.
*/
function advanceCursor(
cursor: SyncCursor,
domain: SyncDomain,
page: Array<{ updatedAt: Date; id: string }>,
): void {
const last = page[page.length - 1];
if (last) {
cursor[domain] = { updatedAtMs: last.updatedAt.getTime(), id: last.id };
}
}
/**
* Whether the cursor can no longer be caught up incrementally, and why.
*
* Past retention: ANY domain watermark older than the tombstone-retention
* horizon. Tombstones older than that may have been pruned by the cleanup
* job, so an incremental delta could silently miss a deletion.
*
* Before a reset: 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. 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.
*/
async function expiry(
userId: string,
token: string | undefined,
cursor: SyncCursor,
serverNow: Date,
): Promise<"retention" | "sync_reset" | null> {
const retentionHorizon = new Date(
serverNow.getTime() - TOMBSTONE_RETENTION_DAYS * DAY_MS,
);
const pastRetention = SYNC_DOMAINS.some((domain) => {
const wm = cursor[domain];
return wm !== undefined && wm.updatedAtMs < retentionHorizon.getTime();
});
let beforeReset = false;
if (token) {
const account = await prisma.user.findUnique({
where: { id: userId },
select: { syncResetAt: true },
});
if (account?.syncResetAt) {
const issuedAt = decodeCursorIssuedAt(token);
beforeReset =
issuedAt === null || issuedAt < account.syncResetAt.getTime();
}
}
if (!pastRetention && !beforeReset) return null;
return beforeReset ? "sync_reset" : "retention";
}
function emptyChanges(): SyncChangesBody["changes"] {
return {
measurements: { upserts: [], tombstones: [] },
mood: { upserts: [], tombstones: [] },
intakes: { upserts: [], tombstones: [] },
cycleDays: { upserts: [], tombstones: [] },
cycles: { upserts: [], tombstones: [] },
};
}
export async function pullSyncChanges(input: {
userId: string;
/** The opaque cursor the client echoed, if any. */
token: string | undefined;
limit: number;
serverNow: Date;
}): Promise<SyncChangesBody> {
const { userId, token, limit, serverNow } = input;
// Module gate: cycle. The `cycleDays` / `cycles` domains carry period
// days and per-day symptom rows, the same data every `/api/cycle/*` route
// refuses with a 403 when the module is off. Without this the delta feed
// re-served them ungated to the very same token.
//
// OMIT, NOT 403, deliberately, and unlike the whole-record FHIR export.
// This feed multiplexes five domains over one cursor: refusing the whole
// request would wedge measurement / mood / intake catch-up for every
// account with cycle tracking off, which (the gate delegates to
// `isCycleEnabled`) is the default for most accounts. A partial payload is
// the only outcome that keeps the client syncing while still honouring the
// promise that a disabled module hands out nothing.
//
// The two cycle watermarks are deliberately NOT advanced while the module
// is off (the queries are skipped, so `advanceCursor` sees an empty page
// and leaves the prior watermark untouched). Re-enabling therefore resumes
// from where the client actually left off instead of silently skipping the
// rows that changed while the module was hidden.
//
// KNOWN GAP, out of scope here: a client that synced cycle rows BEFORE the
// module was turned off keeps its stale local mirror. Emitting tombstones
// for them would be wrong (a tombstone means deleted, and disabling a
// module must leave the rows intact so re-enabling finds a complete
// history). Purging a paired client's mirror needs an explicit "module went
// dark" signal in the feed contract plus client support for it; that is an
// additive wire change and an iOS-client change, not a server-side gate
// fix, so it is left for a coordinated contract cycle.
const cycleEnabled = await isModuleEnabled(userId, "cycle");
// A garbage / unparseable / old-format cursor decodes to null → fresh
// initial sync (every domain scans from the start).
const cursor: SyncCursor = token ? (decodeCursor(token) ?? {}) : {};
const expired = await expiry(userId, token, cursor, serverNow);
if (expired) {
annotate({
action: { name: "sync.changes.pull" },
meta: {
cursor_expired: true,
cursor_expired_reason: expired,
returned: 0,
},
});
return {
serverNow: serverNow.toISOString(),
cursor: token ?? null,
hasMore: false,
cursorExpired: true,
changes: emptyChanges(),
};
}
const rows = await readDomainRows(userId, cursor, limit, cycleEnabled);
const nextCursor: SyncCursor = { ...cursor };
const m = pageOf(rows.measurements, limit);
const measurements = await measurementBlock(userId, m.page);
advanceCursor(nextCursor, "measurements", m.page);
const md = pageOf(rows.mood, limit);
const mood = moodBlock(md.page);
advanceCursor(nextCursor, "mood", md.page);
const ik = pageOf(rows.intakes, limit);
const intakes = intakeBlock(ik.page);
advanceCursor(nextCursor, "intakes", ik.page);
const cd = pageOf(rows.cycleDays, limit);
const cycleDays = cycleDayBlock(cd.page);
advanceCursor(nextCursor, "cycleDays", cd.page);
const cy = pageOf(rows.cycles, limit);
const cycles = cycleBlock(cy.page);
advanceCursor(nextCursor, "cycles", cy.page);
const hasMore =
m.hasMore || md.hasMore || ik.hasMore || cd.hasMore || cy.hasMore;
annotate({
action: { name: "sync.changes.pull" },
meta: {
measurement_upserts: measurements.upserts.length,
measurement_tombstones: measurements.tombstones.length,
mood_upserts: mood.upserts.length,
mood_tombstones: mood.tombstones.length,
intake_upserts: intakes.upserts.length,
intake_tombstones: intakes.tombstones.length,
cycle_day_upserts: cycleDays.upserts.length,
cycle_day_tombstones: cycleDays.tombstones.length,
cycle_upserts: cycles.upserts.length,
cycle_tombstones: cycles.tombstones.length,
cycle_module_enabled: cycleEnabled,
has_more: hasMore,
cursor_present: Boolean(token),
},
});
return {
serverNow: serverNow.toISOString(),
cursor: encodeCursor(nextCursor, serverNow.getTime()),
hasMore,
cursorExpired: false,
changes: { measurements, mood, intakes, cycleDays, cycles },
};
}
+147
View File
@@ -0,0 +1,147 @@
/**
* The per-domain keyset walks of the delta feed.
*
* Live and soft-deleted rows come out of the same scan: a soft-delete bumps
* `updatedAt`, so a tombstone is just a row whose `deletedAt` is non-null.
* Each walk fetches `limit + 1` rows to detect `hasMore` without a count
* query.
*/
import { prisma } from "@/lib/db";
import type { Prisma } from "@/generated/prisma/client";
import type { DomainWatermark, SyncCursor } from "@/lib/sync/cursor";
import { dayLogSymptomInclude } from "@/lib/cycle/dto";
const MEASUREMENT_SELECT = {
id: true,
externalId: true,
type: true,
value: true,
unit: true,
measuredAt: true,
source: true,
notes: true,
notesEncrypted: true,
syncVersion: true,
deletedAt: true,
updatedAt: true,
} satisfies Prisma.MeasurementSelect;
const MOOD_SELECT = {
id: true,
date: true,
mood: true,
score: true,
moodA1: true,
stressA2: true,
energyA3: true,
connectionA4: true,
stabilityA5: true,
context: true,
tags: true,
note: true,
noteEncrypted: true,
moodLoggedAt: true,
source: true,
syncVersion: true,
deletedAt: true,
updatedAt: true,
} satisfies Prisma.MoodEntrySelect;
const INTAKE_SELECT = {
id: true,
medicationId: true,
scheduledFor: true,
takenAt: true,
skipped: true,
source: true,
autoRecorded: true,
syncVersion: true,
deletedAt: true,
updatedAt: true,
} satisfies Prisma.MedicationIntakeEventSelect;
export type MeasurementRow = Prisma.MeasurementGetPayload<{
select: typeof MEASUREMENT_SELECT;
}>;
export type MoodRow = Prisma.MoodEntryGetPayload<{
select: typeof MOOD_SELECT;
}>;
export type IntakeRow = Prisma.MedicationIntakeEventGetPayload<{
select: typeof INTAKE_SELECT;
}>;
export type CycleDayRow = Prisma.CycleDayLogGetPayload<{
include: typeof dayLogSymptomInclude;
}>;
export type CycleRow = Prisma.MenstrualCycleGetPayload<object>;
/**
* Build the `(updatedAt, id)` keyset predicate for a domain's watermark.
* Empty object (full scan) when the client has no watermark for it yet.
*/
function keysetFilter(wm: DomainWatermark | undefined) {
if (!wm) return {};
const at = new Date(wm.updatedAtMs);
return {
OR: [
{ updatedAt: { gt: at } },
{ AND: [{ updatedAt: at }, { id: { gt: wm.id } }] },
],
};
}
export interface DomainRows {
measurements: MeasurementRow[];
mood: MoodRow[];
intakes: IntakeRow[];
cycleDays: CycleDayRow[];
cycles: CycleRow[];
}
/**
* Read one page plus one row of every domain past its watermark. The cycle
* walks are skipped outright when the module is off, not filtered after the
* fact, so the disabled rows are never read out of Postgres at all.
*/
export async function readDomainRows(
userId: string,
cursor: SyncCursor,
limit: number,
cycleEnabled: boolean,
): Promise<DomainRows> {
const [measurements, mood, intakes, cycleDays, cycles] = await Promise.all([
prisma.measurement.findMany({
where: { userId, ...keysetFilter(cursor.measurements) },
orderBy: [{ updatedAt: "asc" }, { id: "asc" }],
take: limit + 1,
select: MEASUREMENT_SELECT,
}),
prisma.moodEntry.findMany({
where: { userId, ...keysetFilter(cursor.mood) },
orderBy: [{ updatedAt: "asc" }, { id: "asc" }],
take: limit + 1,
select: MOOD_SELECT,
}),
prisma.medicationIntakeEvent.findMany({
where: { userId, ...keysetFilter(cursor.intakes) },
orderBy: [{ updatedAt: "asc" }, { id: "asc" }],
take: limit + 1,
select: INTAKE_SELECT,
}),
cycleEnabled
? prisma.cycleDayLog.findMany({
where: { userId, ...keysetFilter(cursor.cycleDays) },
orderBy: [{ updatedAt: "asc" }, { id: "asc" }],
take: limit + 1,
include: dayLogSymptomInclude,
})
: [],
cycleEnabled
? prisma.menstrualCycle.findMany({
where: { userId, ...keysetFilter(cursor.cycles) },
orderBy: [{ updatedAt: "asc" }, { id: "asc" }],
take: limit + 1,
})
: [],
]);
return { measurements, mood, intakes, cycleDays, cycles };
}
+114
View File
@@ -0,0 +1,114 @@
/**
* The wire rows of `GET /api/sync/changes`, one upsert and one tombstone
* shape per domain. Tombstone identity follows the iOS §7.3 table:
* measurements key on `externalId` (the cross-device key); mood and intakes
* key on the server `id` (the client mirrors those by server id); cycle days
* key on `externalId` when present (HealthKit-origin rows are
* externalId-keyed cross-device, like measurements) else the server `id`;
* cycles key on the server `id` (computed rows carry no externalId).
*/
import type { MoodContextWire } from "@/lib/mood/context";
import type { CycleDayLogDTO, MenstrualCycleDTO } from "@/lib/cycle/dto";
export interface MeasurementUpsert {
id: string;
externalId: string | null;
type: string;
value: number;
unit: string;
measuredAt: string;
source: string;
notes: string | null;
syncVersion: number;
updatedAt: string;
}
export interface MeasurementTombstone {
id: string;
externalId: string | null;
syncVersion: number;
deletedAt: string;
updatedAt: string;
}
export interface MoodUpsert {
id: string;
date: string;
mood: string;
score: number;
/**
* The five level-A self-state values, each 0-10 and each nullable. `a2` is
* inverse-oriented: a higher number means more stress, and a client that
* wants "up is better" flips the sign itself rather than the server
* storing a flipped value.
*
* Server-authoritative: `a1` is derived from the five-point label on every
* write, so a client that sends only `mood` still mirrors a complete row.
*/
a1: number | null;
a2: number | null;
a3: number | null;
a4: number | null;
a5: number | null;
/**
* The day's context, or null when the entry carries none. Null and an
* object of nulls are different answers and the feed keeps them apart.
*/
context: MoodContextWire | null;
tags: string | null;
note: string | null;
moodLoggedAt: string;
source: string;
syncVersion: number;
updatedAt: string;
}
/** The tombstone of a domain keyed on the server id alone. */
export interface IdTombstone {
id: string;
syncVersion: number;
deletedAt: string;
updatedAt: string;
}
export interface IntakeUpsert {
id: string;
medicationId: string;
scheduledFor: string;
takenAt: string | null;
skipped: boolean;
source: string;
/** Recorded by the server (`autoMarkTaken`), not the person. */
autoRecorded: boolean;
syncVersion: number;
updatedAt: string;
}
export interface CycleDayTombstone {
id: string;
externalId: string | null;
syncVersion: number;
deletedAt: string;
updatedAt: string;
}
export interface DomainBlock<U, T> {
upserts: U[];
tombstones: T[];
}
export interface SyncChanges {
measurements: DomainBlock<MeasurementUpsert, MeasurementTombstone>;
mood: DomainBlock<MoodUpsert, IdTombstone>;
intakes: DomainBlock<IntakeUpsert, IdTombstone>;
cycleDays: DomainBlock<CycleDayLogDTO, CycleDayTombstone>;
cycles: DomainBlock<MenstrualCycleDTO, IdTombstone>;
}
export interface SyncChangesBody {
serverNow: string;
cursor: string | null;
hasMore: boolean;
cursorExpired: boolean;
changes: SyncChanges;
}