feat(measurements): resolve each reading's Apple Health write-back

A reading sent with a measurements:write token keeps its EXTERNAL source
and now points at the token that carried it (measurements.ingest_token_id).
Every stored row answers a server-resolved appleHealthMirror: true for
WITHINGS, IMPORT and MANUAL, the token's choice for EXTERNAL, false
otherwise. The plain list accepts appleHealthMirror=true so a client can
page exactly those rows.

Migration 0476 adds the columns and links existing EXTERNAL readings only
where one measurement token provably sent them; 0477 builds the index
concurrently. Backups carry the link and a restore keeps it only for a
token the account still holds. A guard holds every EXTERNAL writer to
setting the link.
This commit is contained in:
Marc-André Bombeck committed 2026-10-23 19:27:52 +02:00
1 parent a0dce10c1a
commit 8d10e5ab94
40 files changed
+1014 -142

No files matched your search

@@ -0,0 +1,84 @@
/**
* Every write that labels a measurement `EXTERNAL` also links it to the token
* that carried it (#1272).
*
* The label and the link answer one question together: whether the HealthLog
* app writes the reading into Apple Health is the choice its person made for
* that token (`apple-health-mirror.ts`). An EXTERNAL row written without the
* link resolves to false forever, whatever the token says, and nothing in the
* write path would notice. So the files that resolve `EXTERNAL_SOURCE` are an
* inventory: a measurement writer must set `ingestTokenId` at least as often
* as it assigns the label, and a file that starts using the constant fails
* here until it is classified.
*
* A tripwire, not a proof: it counts, it does not trace data flow. The
* behaviour is pinned by `tests/integration/apple-health-mirror.test.ts`.
*/
import { describe, expect, it } from "vitest";
import { join } from "node:path";
import {
readSourceText,
stripComments,
walkSourceFiles,
} from "./helpers/source-files";
const SRC = join(process.cwd(), "src");
/** Files that write measurement rows labelled with the constant. */
const MEASUREMENT_WRITERS = [
"lib/measurements/batch-ingest/prepare.ts",
"lib/measurements/create/create-readings.ts",
"lib/measurements/create/create-sleep-stage.ts",
];
/** Files that use the constant for something other than a measurement write. */
const NOT_MEASUREMENT_WRITES: Record<string, string> = {
"app/api/tokens/measurements/route.ts":
"counts the readings that belong to no token (a read)",
"lib/validations/measurement.ts": "the source enum and the edit allowlist",
"lib/workouts/batch-ingest/entries.ts":
"labels workout rows, which carry no token link",
};
function usesOfConstant(rel: string): {
assignments: number;
links: number;
} {
const code = stripComments(readSourceText(join(SRC, rel)));
const assignments = (code.match(/\?\s*EXTERNAL_SOURCE\b/g) ?? []).length;
// A value written under the key, or the shorthand. Not a type member
// (`ingestTokenId: string | null`) and not a read (`writer.ingestTokenId`),
// which write nothing.
const links = (
code.match(/(?<![.\w])ingestTokenId(?:\s*:(?!\s*string\b)|\s*[,}])/g) ?? []
).length;
return { assignments, links };
}
describe("an EXTERNAL measurement carries its token", () => {
const files = walkSourceFiles(SRC, { floor: 3000 })
.filter((p) => !p.startsWith("generated/"))
.filter((p) => !p.includes("__tests__"))
.filter((p) => !/\.test\.tsx?$/.test(p))
.filter((p) => p !== "lib/measurements/external-source.ts")
.filter((p) =>
/\bEXTERNAL_SOURCE\b/.test(stripComments(readSourceText(join(SRC, p)))),
);
it("knows every file that uses the constant", () => {
expect(files.length).toBeGreaterThan(0);
expect(files).toEqual(
[...MEASUREMENT_WRITERS, ...Object.keys(NOT_MEASUREMENT_WRITES)].sort(),
);
});
it.each(MEASUREMENT_WRITERS)(
"%s links the token wherever it assigns the label",
(rel) => {
const { assignments, links } = usesOfConstant(rel);
expect(assignments).toBeGreaterThan(0);
expect(links).toBeGreaterThanOrEqual(assignments);
},
);
});
+81 -59
View File
@@ -17,9 +17,13 @@ import {
validateMeasurementRange,
} from "@/lib/validations/measurement";
import { encryptNote, shapeMeasurementNotes } from "@/lib/crypto/note-cipher";
import {
APPLE_HEALTH_MIRROR_INCLUDE,
withAppleHealthMirror,
} from "@/lib/measurements/apple-health-mirror";
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation";
import { Prisma } from "@/generated/prisma/client";
import { Prisma, type MeasurementType } from "@/generated/prisma/client";
import { NextRequest } from "next/server";
import { z } from "zod";
@@ -37,6 +41,7 @@ export const GET = apiHandler(
// is not part of a unique index.
const measurement = await prisma.measurement.findFirst({
where: { id, deletedAt: null },
include: APPLE_HEALTH_MIRROR_INCLUDE,
});
if (!measurement || measurement.userId !== user.id) {
@@ -51,7 +56,9 @@ export const GET = apiHandler(
},
});
return apiSuccess(shapeMeasurementNotes(measurement));
return apiSuccess(
withAppleHealthMirror(shapeMeasurementNotes(measurement)),
);
},
);
@@ -113,62 +120,8 @@ export const PUT = apiHandler(
// implausible edited value flowed unchecked into rollups, the health
// score, the BP gates and the Coach snapshot.
if (data.value !== undefined && data.value !== existing.value) {
// Server-owned rows first: a value attributed to a connector / import /
// computed engine is the provider's reading — editing the number would
// forge a source-attributed row the server never received. Timestamp
// and note edits stay allowed — annotating a Withings reading is
// legitimate.
//
// v1.38.x — gated on `USER_CORRECTABLE_MEASUREMENT_SOURCES` rather than
// the write-side `WRITABLE_MEASUREMENT_SOURCES` the two once shared,
// because `EXTERNAL` separates the questions. A client may not NAME that
// source (it is resolved from the ingest credential, and the write
// allowlist is published as `ingest.writeAllowlist`), but the hardware
// behind the token is the user's own scale, so the rationale above —
// "the value is the provider's" — does not reach it. Locking those rows
// would also break the correctability the settings card and the Home
// Assistant guide promise. This is the only site that reads the wider
// set; re-merging the two constants silently value-locks bridge rows.
if (
!(USER_CORRECTABLE_MEASUREMENT_SOURCES as readonly string[]).includes(
existing.source,
)
) {
annotate({
action: { name: "measurements.update.server-owned-source" },
meta: { measurement_id: id, source: existing.source },
});
return apiError(
"Values from a connected source cannot be edited",
409,
{ errorCode: "measurement.update.server_owned_source" },
);
}
// Range check against the row's OWN type (the edit body carries no
// type). Returned through the standard multi-issue 422 envelope so the
// edit sheet renders it like any other field error.
const rangeCheck = z
.object({
value: z.number().superRefine((value, ctx) => {
const rangeError = validateMeasurementRange(existing.type, value);
if (rangeError) {
ctx.addIssue({ code: "custom", message: rangeError });
}
}),
})
.safeParse({ value: data.value });
if (!rangeCheck.success) {
annotate({
action: { name: "measurements.update.validation-failed" },
meta: {
issue_count: rangeCheck.error.issues.length,
measurement_id: id,
reason: "value_out_of_range",
},
});
return returnAllZodIssues(rangeCheck.error, 422);
}
const refused = refuseValueEdit(existing, data.value, id);
if (refused) return refused;
}
let measurement;
@@ -185,6 +138,7 @@ export const PUT = apiHandler(
notesEncrypted: encryptNote(data.notes),
}),
},
include: APPLE_HEALTH_MIRROR_INCLUDE,
});
} catch (err) {
// v1.4.28 — re-pointing `measuredAt` onto an existing
@@ -252,7 +206,9 @@ export const PUT = apiHandler(
{ source: "manual" },
);
return apiSuccess(shapeMeasurementNotes(measurement));
return apiSuccess(
withAppleHealthMirror(shapeMeasurementNotes(measurement)),
);
},
);
@@ -320,3 +276,69 @@ export const DELETE = apiHandler(
return apiSuccess({ deleted: true });
},
);
/**
* Whether a changed value may be written over `existing`, as the refusal to
* answer when it may not; null when it may.
*/
function refuseValueEdit(
existing: { source: string; type: MeasurementType },
value: number,
id: string,
): Response | null {
// Server-owned rows first: a value attributed to a connector / import /
// computed engine is the provider's reading — editing the number would
// forge a source-attributed row the server never received. Timestamp
// and note edits stay allowed — annotating a Withings reading is
// legitimate.
//
// v1.38.x — gated on `USER_CORRECTABLE_MEASUREMENT_SOURCES` rather than
// the write-side `WRITABLE_MEASUREMENT_SOURCES` the two once shared,
// because `EXTERNAL` separates the questions. A client may not NAME that
// source (it is resolved from the ingest credential, and the write
// allowlist is published as `ingest.writeAllowlist`), but the hardware
// behind the token is the user's own scale, so the rationale above —
// "the value is the provider's" — does not reach it. Locking those rows
// would also break the correctability the settings card and the Home
// Assistant guide promise. This is the only site that reads the wider
// set; re-merging the two constants silently value-locks bridge rows.
if (
!(USER_CORRECTABLE_MEASUREMENT_SOURCES as readonly string[]).includes(
existing.source,
)
) {
annotate({
action: { name: "measurements.update.server-owned-source" },
meta: { measurement_id: id, source: existing.source },
});
return apiError("Values from a connected source cannot be edited", 409, {
errorCode: "measurement.update.server_owned_source",
});
}
// Range check against the row's OWN type (the edit body carries no
// type). Returned through the standard multi-issue 422 envelope so the
// edit sheet renders it like any other field error.
const rangeCheck = z
.object({
value: z.number().superRefine((value, ctx) => {
const rangeError = validateMeasurementRange(existing.type, value);
if (rangeError) {
ctx.addIssue({ code: "custom", message: rangeError });
}
}),
})
.safeParse({ value: value });
if (!rangeCheck.success) {
annotate({
action: { name: "measurements.update.validation-failed" },
meta: {
issue_count: rangeCheck.error.issues.length,
measurement_id: id,
reason: "value_out_of_range",
},
});
return returnAllZodIssues(rangeCheck.error, 422);
}
return null;
}
+2
View File
@@ -261,6 +261,8 @@ async function postBatch(request: NextRequest): Promise<Response> {
const result = await ingestMeasurementBatch({
user: { id: user.id, timezone: user.timezone },
scoped,
// On the Bearer path the token's id rides `session.id`.
ingestTokenId: scoped ? auth.session.id : null,
entries,
syncTrigger: parsed.data.syncTrigger,
ipAddress: getClientIp(request),
+14 -2
View File
@@ -21,6 +21,10 @@ import {
} from "@/lib/validations/measurement";
import { withIdempotency } from "@/lib/idempotency";
import { shapeMeasurementNotes } from "@/lib/crypto/note-cipher";
import {
type MeasurementWithMirrorInput,
withAppleHealthMirror,
} from "@/lib/measurements/apple-health-mirror";
import { measurementListResponse } from "@/lib/measurements/list/measurement-list";
import {
createReading,
@@ -154,6 +158,8 @@ async function postMeasurement(request: NextRequest) {
const writer: ReadingWriter = {
user: { id: user.id, timezone: user.timezone },
scoped,
// On the Bearer path the token's id rides `session.id`.
ingestTokenId: scoped ? auth.session.id : null,
ipAddress: getClientIp(request),
};
@@ -202,7 +208,7 @@ async function postMeasurement(request: NextRequest) {
if ("duplicate" in created) {
return apiError(DUPLICATE_MESSAGE, 409, DUPLICATE_META);
}
return apiSuccess(created.rows.map(shapeMeasurementNotes), 201);
return apiSuccess(created.rows.map(answerRow), 201);
}
// Single mode (existing behavior)
@@ -237,7 +243,12 @@ async function postMeasurement(request: NextRequest) {
if ("duplicate" in created) {
return apiError(DUPLICATE_MESSAGE, 409, DUPLICATE_META);
}
return apiSuccess(shapeMeasurementNotes(created.row), 201);
return apiSuccess(answerRow(created.row), 201);
}
/** A written row as it is answered: the note readable, the write-back resolved. */
function answerRow(row: MeasurementWithMirrorInput) {
return withAppleHealthMirror(shapeMeasurementNotes(row));
}
/**
@@ -262,6 +273,7 @@ async function postNamedSleepStage(
{
userId: writer.user.id,
scoped: writer.scoped,
ingestTokenId: writer.ingestTokenId,
ipAddress: writer.ipAddress,
},
parsed.data,
@@ -66,6 +66,8 @@ export interface MeasurementInsertRow {
sleepStage: string | null;
rhythmClassification: string | null;
deviceType: string | null;
/** The token an EXTERNAL row came from (#1272); null when absent. */
ingestTokenId?: string | null;
syncVersion: number;
deletedAt: Date | null;
createdAt?: Date;
@@ -98,6 +100,7 @@ const COLUMNS: { [K in keyof MeasurementInsertRow]-?: [string, Kind] } = {
sleepStage: ["sleep_stage", "enum:sleep_stage"],
rhythmClassification: ["rhythm_classification", "enum:rhythm_classification"],
deviceType: ["device_type", "text"],
ingestTokenId: ["ingest_token_id", "text"],
syncVersion: ["sync_version", "int4"],
deletedAt: ["deleted_at", "timestamp"],
createdAt: ["created_at", "timestamp"],
@@ -41,6 +41,9 @@ export const measurementSchema = z
sleepStage: z.enum(SleepStage).nullable().optional(),
rhythmClassification: z.enum(RhythmClassification).nullable().optional(),
deviceType: z.string().nullable().optional(),
// The token that carried an EXTERNAL reading (#1272). Kept on restore
// only when the restoring account still holds that token.
ingestTokenId: z.string().max(64).nullable().optional(),
syncVersion: z.number().int().optional(),
deletedAt: isoDateTime.nullable().optional(),
createdAt: isoDateTime.optional(),
+2
View File
@@ -42,6 +42,7 @@ const MEASUREMENT_BACKUP_SELECT = {
sleepStage: true,
rhythmClassification: true,
deviceType: true,
ingestTokenId: true,
syncVersion: true,
deletedAt: true,
createdAt: true,
@@ -86,6 +87,7 @@ function serialiseMeasurement(
sleepStage: measurement.sleepStage,
rhythmClassification: measurement.rhythmClassification,
deviceType: measurement.deviceType,
ingestTokenId: measurement.ingestTokenId,
syncVersion: measurement.syncVersion,
deletedAt: measurement.deletedAt?.toISOString() ?? null,
createdAt: measurement.createdAt.toISOString(),
+25 -4
View File
@@ -156,7 +156,10 @@ export async function clearMeasurementRollups(
await ctx.tx.measurementRollup.deleteMany({ where: { userId: ctx.ownerId } });
}
function toRestoredMeasurementData(measurement: BackupMeasurement) {
function toRestoredMeasurementData(
measurement: BackupMeasurement,
ownTokenIds: ReadonlySet<string>,
) {
return {
type: isLegacyRmssdRow(measurement)
? ("HRV_RMSSD" as const)
@@ -204,6 +207,13 @@ function toRestoredMeasurementData(measurement: BackupMeasurement) {
sleepStage: (measurement.sleepStage ?? null) as never,
rhythmClassification: (measurement.rhythmClassification ?? null) as never,
deviceType: measurement.deviceType ?? null,
// The token link (#1272) survives only onto the account that still holds
// the token: a backup restored elsewhere, or after the token was deleted,
// would point at a row that is not there or not this account's.
ingestTokenId:
measurement.ingestTokenId && ownTokenIds.has(measurement.ingestTokenId)
? measurement.ingestTokenId
: null,
syncVersion: measurement.syncVersion ?? 1,
deletedAt: measurement.deletedAt ? new Date(measurement.deletedAt) : null,
...(measurement.createdAt
@@ -343,8 +353,9 @@ async function reconcileWithoutId(
tx: Prisma.TransactionClient,
ownerId: string,
measurement: BackupMeasurement,
ownTokenIds: ReadonlySet<string>,
): Promise<void> {
const restoredData = toRestoredMeasurementData(measurement);
const restoredData = toRestoredMeasurementData(measurement, ownTokenIds);
const existing = await tx.measurement.findFirst({
where: {
userId: ownerId,
@@ -378,6 +389,16 @@ export async function restoreMeasurements(ctx: RestoreContext): Promise<void> {
const measurementBatchSize = 1_000;
const isExpiredTombstone = await expiredTombstoneFilter(ctx);
const { isShadowedLegacyRmssd } = ctx.measurementPrepass;
// Tokens are not part of a backup; the ones the account holds now decide
// which token links come back.
const ownTokenIds = new Set(
(
await tx.apiToken.findMany({
where: { userId: ownerId },
select: { id: true },
})
).map((token) => token.id),
);
ctx.reportMeasurements(0);
await streamed.forEachMeasurementBatch(
measurementBatchSize,
@@ -395,7 +416,7 @@ export async function restoreMeasurements(ctx: RestoreContext): Promise<void> {
{
id: measurement.id,
userId: ownerId,
...toRestoredMeasurementData(measurement),
...toRestoredMeasurementData(measurement, ownTokenIds),
},
]
: [],
@@ -412,7 +433,7 @@ export async function restoreMeasurements(ctx: RestoreContext): Promise<void> {
}
for (const measurement of batch) {
if (measurement.id) continue;
await reconcileWithoutId(tx, ownerId, measurement);
await reconcileWithoutId(tx, ownerId, measurement, ownTokenIds);
}
},
);
@@ -0,0 +1,84 @@
/**
* The Apple Health write-back rule (#1272), over every source the enum has.
*
* WITHINGS, IMPORT and MANUAL are true whatever the row's token says, the
* app's own rule as it stood; EXTERNAL follows its token; everything else is
* false. The Prisma filter the list's `appleHealthMirror=true` uses is the
* same rule written as a `where`, checked here against the resolver for each
* source and token state (the database agrees in
* `tests/integration/apple-health-mirror.test.ts`).
*/
import { describe, expect, it } from "vitest";
import { MEASUREMENT_SOURCES } from "@/lib/measurements/sources";
import {
APPLE_HEALTH_MIRROR_WHERE,
resolveAppleHealthMirror,
withAppleHealthMirror,
} from "../apple-health-mirror";
const TOKEN_STATES = [
undefined,
null,
{ mirrorToAppleHealth: false },
{ mirrorToAppleHealth: true },
] as const;
/** The filter's two arms, evaluated against one row. */
function matchesWhere(row: {
source: string;
ingestToken?: { mirrorToAppleHealth: boolean } | null;
}): boolean {
const [always, external] = APPLE_HEALTH_MIRROR_WHERE.OR;
return (
(always.source.in as readonly string[]).includes(row.source) ||
(row.source === external.source &&
row.ingestToken?.mirrorToAppleHealth ===
external.ingestToken.mirrorToAppleHealth)
);
}
describe("resolveAppleHealthMirror", () => {
it("is true for WITHINGS, IMPORT and MANUAL and an opted-in EXTERNAL", () => {
const truthy = MEASUREMENT_SOURCES.flatMap((source) =>
TOKEN_STATES.filter((ingestToken) =>
resolveAppleHealthMirror({ source, ingestToken }),
).map((ingestToken) => `${source}:${JSON.stringify(ingestToken)}`),
);
expect(truthy.sort()).toEqual(
[
...["IMPORT", "MANUAL", "WITHINGS"].flatMap((source) =>
TOKEN_STATES.map((s) => `${source}:${JSON.stringify(s)}`),
),
'EXTERNAL:{"mirrorToAppleHealth":true}',
].sort(),
);
});
it("agrees with the list filter for every source and token state", () => {
for (const source of MEASUREMENT_SOURCES) {
for (const ingestToken of TOKEN_STATES) {
expect(
matchesWhere({ source, ingestToken }),
`${source} ${JSON.stringify(ingestToken)}`,
).toBe(resolveAppleHealthMirror({ source, ingestToken }));
}
}
});
});
describe("withAppleHealthMirror", () => {
it("adds the decision and drops the token and its id", () => {
const row = withAppleHealthMirror({
id: "m1",
source: "EXTERNAL",
ingestTokenId: "tok-1",
ingestToken: { mirrorToAppleHealth: true },
});
expect(row).toEqual({
id: "m1",
source: "EXTERNAL",
appleHealthMirror: true,
});
});
});
@@ -0,0 +1,96 @@
/**
* Whether the HealthLog app writes a server measurement into Apple Health
* (#1272). Resolved here, once, and published on every measurement row as
* `appleHealthMirror`; the app mirrors exactly the rows that say true instead
* of deriving the decision from `source`.
*
* The rule:
*
* - WITHINGS, IMPORT and MANUAL: true. This is the app's own rule as it
* stands (`MeasurementSource.isServerMirrorEligible`), so publishing the
* field changes nothing for an existing row. MANUAL is deliberate: a
* reading typed on the web or on another device is stored as MANUAL and
* never reached Apple Health otherwise, and a manual row the app already
* wrote at create time is recognised by the app's own dedup (the sample
* carries the server id), not by its source.
* - EXTERNAL: what the person chose for the `measurements:write` token that
* carried the reading (`ApiToken.mirrorToAppleHealth`). The source stays
* EXTERNAL either way; a token never claims another provenance. A row with
* no token (written before the link existed and not attributable to one)
* is not mirrored.
* - Everything else (APPLE_HEALTH, the provider syncs, COMPUTED, TELEGRAM,
* MCP, …): false. Apple Health rows came from Apple Health, and the
* provider and server-written rows are never authored into it.
*/
import type { MeasurementSource, Prisma } from "@/generated/prisma/client";
/** Sources the app mirrors whatever the token says. */
const ALWAYS_MIRRORED: ReadonlySet<MeasurementSource> = new Set([
"WITHINGS",
"IMPORT",
"MANUAL",
]);
/** What a row read needs to resolve the decision. */
export const APPLE_HEALTH_MIRROR_SELECT = {
source: true,
ingestToken: { select: { mirrorToAppleHealth: true } },
} as const satisfies Prisma.MeasurementSelect;
/** The same, for a read or write that returns the whole row. */
export const APPLE_HEALTH_MIRROR_INCLUDE = {
ingestToken: APPLE_HEALTH_MIRROR_SELECT.ingestToken,
} as const satisfies Prisma.MeasurementInclude;
/** A whole row read or written with `APPLE_HEALTH_MIRROR_INCLUDE`. */
export type MeasurementWithMirrorInput = Prisma.MeasurementGetPayload<{
include: typeof APPLE_HEALTH_MIRROR_INCLUDE;
}>;
export function resolveAppleHealthMirror(row: {
source: MeasurementSource | string;
ingestToken?: { mirrorToAppleHealth: boolean } | null;
}): boolean {
if (ALWAYS_MIRRORED.has(row.source as MeasurementSource)) return true;
if (row.source === "EXTERNAL") {
return row.ingestToken?.mirrorToAppleHealth === true;
}
return false;
}
/**
* The rows that resolve true, as a Prisma filter: the list's
* `appleHealthMirror=true` filter, which the app's historical backfill pages
* instead of one fetch per source. Kept beside the rule so the two cannot
* disagree (pinned by `apple-health-mirror.test.ts`).
*/
export const APPLE_HEALTH_MIRROR_WHERE = {
OR: [
{ source: { in: [...ALWAYS_MIRRORED] } },
{ source: "EXTERNAL", ingestToken: { mirrorToAppleHealth: true } },
],
} as const satisfies Prisma.MeasurementWhereInput;
/**
* A row read with `APPLE_HEALTH_MIRROR_SELECT` (or `_INCLUDE`), answered: the
* resolved `appleHealthMirror` added, the token relation and the token id
* dropped, so neither ever reaches the wire.
*/
export function withAppleHealthMirror<
T extends {
source: MeasurementSource | string;
ingestToken?: { mirrorToAppleHealth: boolean } | null;
ingestTokenId?: string | null;
},
>(
row: T,
): Omit<T, "ingestToken" | "ingestTokenId"> & { appleHealthMirror: boolean } {
const { ingestToken, ingestTokenId: _ingestTokenId, ...rest } = row;
return {
...rest,
appleHealthMirror: resolveAppleHealthMirror({
source: row.source,
ingestToken,
}),
};
}
@@ -34,6 +34,11 @@ export interface BatchIngestInput {
user: { id: string; timezone: string | null };
/** A narrow single-purpose Bearer (a bridge), not the native client. */
scoped: boolean;
/**
* The bridge token that carried the batch, set exactly when `scoped`: each
* row it writes points at it (`Measurement.ingestTokenId`, #1272).
*/
ingestTokenId: string | null;
entries: BatchIngestEntry[];
syncTrigger: HealthKitRequestSyncTrigger | undefined;
ipAddress: string | null;
+15 -2
View File
@@ -44,9 +44,13 @@ export interface PreparedBatch {
}
export async function prepareBatch(
input: Pick<BatchIngestInput, "user" | "scoped" | "entries">,
input: Pick<
BatchIngestInput,
"user" | "scoped" | "ingestTokenId" | "entries"
>,
): Promise<PreparedBatch> {
const { user, scoped, entries } = input;
const ingestTokenId = scoped ? input.ingestTokenId : null;
const batch: PreparedBatch = {
prepared: [],
results: new Array(entries.length),
@@ -94,7 +98,13 @@ export async function prepareBatch(
continue;
}
const prepared = prepareMeasurementEntry(index, entry, user.id, scoped);
const prepared = prepareMeasurementEntry(
index,
entry,
user.id,
scoped,
ingestTokenId,
);
if ("row" in prepared) batch.prepared.push(prepared);
else batch.results[index] = prepared;
}
@@ -183,6 +193,7 @@ function prepareMeasurementEntry(
entry: BatchIngestEntry,
userId: string,
scoped: boolean,
ingestTokenId: string | null,
): PreparedEntry | EntryResult {
const mapped = mapAppleHealthEntry({
hkIdentifier: entry.hkIdentifier,
@@ -293,6 +304,8 @@ function prepareMeasurementEntry(
// `(userId, type, EXTERNAL, externalId)`, which is the isolated
// namespace this source exists to give.
source: scoped ? EXTERNAL_SOURCE : (entry.source ?? "APPLE_HEALTH"),
// The token an EXTERNAL row came from (#1272); null for the phone.
ingestTokenId,
measuredAt: mapped.takenAt,
externalId: entry.externalId,
externalSourceVersion: entry.externalSourceVersion ?? null,
@@ -92,6 +92,11 @@ function desiredFor(userId: string, p: PreparedEntry) {
.rhythmClassification as Prisma.MeasurementUncheckedCreateInput["rhythmClassification"],
deviceType: p.row.deviceType as string | null,
aggregationProvenance: statsRow ? ("HEALTHKIT_STATISTICS" as const) : null,
// Only a bridge row names its token (#1272); a phone row leaves the
// column as it is.
...(p.row.ingestTokenId
? { ingestTokenId: p.row.ingestTokenId as string }
: {}),
};
}
+15 -2
View File
@@ -10,6 +10,10 @@ import type { z } from "zod/v4";
import { prisma } from "@/lib/db";
import { EXTERNAL_SOURCE } from "@/lib/measurements/external-source";
import {
APPLE_HEALTH_MIRROR_INCLUDE,
type MeasurementWithMirrorInput,
} from "@/lib/measurements/apple-health-mirror";
import { annotate } from "@/lib/logging/context";
import { fireAndForget } from "@/lib/logging/fire-and-forget";
import { auditLog } from "@/lib/auth/audit";
@@ -39,6 +43,11 @@ export interface ReadingWriter {
user: { id: string; timezone: string | null };
/** A narrow bridge credential: its rows are `EXTERNAL`. */
scoped: boolean;
/**
* The bridge token, set exactly when `scoped`: each row it writes points
* at it (`Measurement.ingestTokenId`, #1272).
*/
ingestTokenId: string | null;
ipAddress: string | null;
}
@@ -52,7 +61,7 @@ export const DUPLICATE_READING = { duplicate: true } as const;
export async function createReadings(
writer: ReadingWriter,
input: z.output<typeof createBatchMeasurementSchema>,
): Promise<{ rows: Measurement[] } | typeof DUPLICATE_READING> {
): Promise<{ rows: MeasurementWithMirrorInput[] } | typeof DUPLICATE_READING> {
const { user, scoped } = writer;
// The caller has to be told which condition the batch hit. Untouched, a
// duplicate escaped as an unhandled error and the batch answered 500,
@@ -77,6 +86,7 @@ export async function createReadings(
source: (scoped
? EXTERNAL_SOURCE
: (m.source ?? "MANUAL")) as MeasurementSource,
ingestTokenId: scoped ? writer.ingestTokenId : null,
measuredAt: m.measuredAt,
// The note is sealed at rest; the legacy plaintext column is
// written null for new rows.
@@ -88,6 +98,7 @@ export async function createReadings(
// `deviceType` instead of silently dropping it.
deviceType: m.deviceType ?? null,
},
include: APPLE_HEALTH_MIRROR_INCLUDE,
}),
),
);
@@ -123,7 +134,7 @@ export async function createReadings(
export async function createReading(
writer: ReadingWriter,
input: z.output<typeof createMeasurementSchema>,
): Promise<{ row: Measurement } | typeof DUPLICATE_READING> {
): Promise<{ row: MeasurementWithMirrorInput } | typeof DUPLICATE_READING> {
const { user, scoped } = writer;
const {
type,
@@ -150,6 +161,7 @@ export async function createReading(
source: (scoped
? EXTERNAL_SOURCE
: (source ?? "MANUAL")) as MeasurementSource,
ingestTokenId: scoped ? writer.ingestTokenId : null,
measuredAt,
// The note is sealed at rest; the legacy plaintext column is nulled.
notes: null,
@@ -160,6 +172,7 @@ export async function createReading(
// silently dropping it.
deviceType: deviceType ?? null,
},
include: APPLE_HEALTH_MIRROR_INCLUDE,
});
} catch (err) {
// `isP2002` rather than `instanceof Prisma.PrismaClientKnownRequestError`.
@@ -23,6 +23,10 @@ import { auditLog } from "@/lib/auth/audit";
import { apiError, apiSuccess } from "@/lib/api-response";
import { getUnitForType } from "@/lib/validations/measurement";
import { shapeMeasurementNotes } from "@/lib/crypto/note-cipher";
import {
APPLE_HEALTH_MIRROR_INCLUDE,
withAppleHealthMirror,
} from "@/lib/measurements/apple-health-mirror";
import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
@@ -51,6 +55,8 @@ export async function createNamedSleepStage(
args: {
userId: string;
scoped: boolean;
/** The bridge token, set exactly when `scoped` (#1272). */
ingestTokenId: string | null;
ipAddress: string | null;
},
entry: z.output<typeof sleepStageEntrySchema>,
@@ -113,6 +119,10 @@ export async function createNamedSleepStage(
externalSourceVersion: null,
sleepStage: segment.stage,
deviceType: entry.deviceType ?? null,
// The token an EXTERNAL row came from (#1272).
...(scoped && args.ingestTokenId
? { ingestTokenId: args.ingestTokenId }
: {}),
},
{ exactExternalMatch: "duplicate" },
),
@@ -190,9 +200,12 @@ async function namedSleepStageResponse(
cross_source_merge: meta.crossSource,
},
});
const row = await prisma.measurement.findUniqueOrThrow({ where: { id } });
const row = await prisma.measurement.findUniqueOrThrow({
where: { id },
include: APPLE_HEALTH_MIRROR_INCLUDE,
});
return apiSuccess(
{ ...shapeMeasurementNotes(row), status },
{ ...withAppleHealthMirror(shapeMeasurementNotes(row)), status },
status === "inserted" ? 201 : 200,
);
}
+24 -3
View File
@@ -20,6 +20,11 @@ import {
import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping";
import { localDayWindow } from "@/lib/tz/local-day";
import { shapeMeasurementNotes } from "@/lib/crypto/note-cipher";
import {
APPLE_HEALTH_MIRROR_SELECT,
APPLE_HEALTH_MIRROR_WHERE,
withAppleHealthMirror,
} from "@/lib/measurements/apple-health-mirror";
import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
import {
readDailySeries,
@@ -59,8 +64,18 @@ const MEASUREMENT_WIRE_SELECT = {
deletedAt: true,
createdAt: true,
updatedAt: true,
// Resolved onto `appleHealthMirror` and dropped by `withAppleHealthMirror`.
ingestToken: APPLE_HEALTH_MIRROR_SELECT.ingestToken,
} as const satisfies Prisma.MeasurementSelect;
/** One row as it is answered: the note readable, the write-back resolved. */
function wireRow<
T extends Parameters<typeof shapeMeasurementNotes>[0] &
Parameters<typeof withAppleHealthMirror>[0],
>(row: T) {
return withAppleHealthMirror(shapeMeasurementNotes(row));
}
type ListQuery = z.output<typeof listMeasurementsSchema>;
type ListUser = { id: string; timezone: string | null };
@@ -202,7 +217,7 @@ async function drillDownResponse(
return apiSuccess({
// The same row shape as the plain page: the note decrypted onto `notes`,
// the ciphertext never on the wire.
measurements: samples.map(shapeMeasurementNotes),
measurements: samples.map(wireRow),
meta: { total: samples.length, limit, offset: 0, dayKey },
});
}
@@ -281,7 +296,7 @@ async function plainPageResponse(
query: ListQuery,
): Promise<Response> {
const { type, from, to, limit, offset, sortBy, sortDir } = query;
const { sourceEq, valueMin, valueMax } = query;
const { sourceEq, valueMin, valueMax, appleHealthMirror } = query;
const where = {
userId,
// Soft-deleted readings are invisible to every list / count / aggregate
@@ -294,6 +309,9 @@ async function plainPageResponse(
// opt-in (which selects the rollup read tier); `sourceEq` narrows the
// plain list to one ingest source.
...(sourceEq && { source: sourceEq }),
// The rows the app mirrors into Apple Health, by the same rule the
// field is resolved with (`apple-health-mirror.ts`).
...(appleHealthMirror === "true" && { AND: [APPLE_HEALTH_MIRROR_WHERE] }),
// Value-range filter. Open-ended on either side; the validator has
// already rejected an inverted `valueMin > valueMax`.
...(valueMin != null || valueMax != null
@@ -330,6 +348,9 @@ async function plainPageResponse(
orderBy: [{ [sortBy]: sortDir }, { id: sortDir }],
take: limit,
skip: offset,
// Named columns, so a column added to the model (a token link) reaches
// the wire only through this select.
select: MEASUREMENT_WIRE_SELECT,
}),
prisma.measurement.count({ where }),
]);
@@ -339,7 +360,7 @@ async function plainPageResponse(
return apiSuccess({
// Decrypt `notesEncrypted` onto `notes` and strip the ciphertext before
// it leaves the server (iOS MeasurementWireDTO contract).
measurements: measurements.map(shapeMeasurementNotes),
measurements: measurements.map(wireRow),
meta: { total, limit, offset },
});
}
+21
View File
@@ -394,6 +394,10 @@ export const listMeasurementsSchema = z
// threaded into the plain-list `where`. Backed by the `(userId, source,
// measuredAt)` index (migration 0136).
sourceEq: measurementSourceEnum.optional(),
// Only the rows the HealthLog app writes into Apple Health
// (`appleHealthMirror`, #1272): the app's historical backfill pages this
// instead of one fetch per source. A plain-list filter.
appleHealthMirror: z.enum(["true"]).optional(),
// v1.18.5 — value-range FILTER (backlog G). Narrows the plain list to
// readings whose `value` falls within an optional `[valueMin, valueMax]`
// band. Either bound may be omitted for an open-ended range. Coerced
@@ -478,6 +482,23 @@ export const listMeasurementsSchema = z
message: "valueMin must be <= valueMax",
path: ["valueMin"],
},
)
// `appleHealthMirror` narrows the plain row list. The other modes answer
// something else (night rows, day buckets, a drill-down, a series), where
// a row filter would either do nothing or change what the mode means.
.refine(
({ appleHealthMirror, type, dayKey, groupBy, aggregate, source }) =>
appleHealthMirror === undefined ||
(type !== "SLEEP_DURATION" &&
dayKey === undefined &&
groupBy === undefined &&
source === undefined &&
(aggregate === undefined || aggregate === "raw")),
{
message:
"appleHealthMirror filters the plain row list only; combine it with type, from, to, sourceEq and paging",
path: ["appleHealthMirror"],
},
);
export const createBatchMeasurementSchema = z.object({