fix(cycle): open cycles from Garmin and Oura period days

Both wearables wrote their period days with a bare create and so skipped
the seam every other cycle writer goes through: a first bleeding day never
opened its cycle and no day was attributed to one. An account whose
periods came from a watch or a ring held day-logs only, with no cycle
start, no lengths and no forecast.

The create now goes through one shared helper that keeps the
create-only-if-absent rule and then runs the same attribution and
`ensureCycleForBleedingDay` inference as the shared upsert. The hourly
legs and the newest-first Garmin backfill both use it; a finished Garmin
period also stamps its last day on a cycle that has no period end yet.

Rows imported before this change are repaired by a one-shot job sent from
boot discovery while candidates exist. It opens or attributes cycles
oldest first per account, touches no day-log content, and converges to an
empty candidate set.

Refs #1333
This commit is contained in:
Marc-André Bombeck committed 2026-11-04 22:02:48 +01:00
1 parent a35b77c147
commit 148af765c9
17 files changed
+901 -131

No files matched your search

+6
View File
@@ -18,6 +18,12 @@
### Fixed
- **Periods from Garmin and Oura open their cycles.** Period days imported
from a watch or a ring were stored as days only, so the cycle tracker saw
no cycle start, no lengths and no forecast. They now open and fill their
cycles like a period logged by hand, a finished Garmin period also marks
its last day, and periods imported earlier are repaired once on the next
start. A day you logged yourself is never changed. (#1333)
- **Typing into the smoking history.** "When you stopped" only took text
that was already a whole YYYY-MM month, so every keystroke was dropped
and only a paste got in. The start year clamped each keystroke into its
+113
View File
@@ -0,0 +1,113 @@
/**
* Period days a wearable reports (Garmin, Oura), written as a SUBORDINATE
* source.
*
* `CycleDayLog` holds one canonical row per `(userId, date)`, and the shared
* upsert (`./day-log-write.ts`) partial-merges a re-post over whatever is
* stored regardless of source. An hourly poll going through it would demote a
* day the person logged themselves (their HEAVY re-tagged LIGHT on the next
* tick). So a wearable day is only ever CREATED, when no row exists for that
* date at all, and never updated.
*
* What the shared upsert does beyond the row itself happens here too, through
* the same functions: a created day is attributed to the cycle it falls in,
* and a first bleeding day opens its cycle (`ensureCycleForBleedingDay`),
* which closes the cycle before it and takes its own end from the one after
* it. Without that a period imported from a watch was a run of rows the
* engine never read: no cycle start, no lengths, no forecast.
*/
import { prisma } from "@/lib/db";
import { getEvent } from "@/lib/logging/context";
import type { FlowLevel } from "@/generated/prisma/client";
import { ensureCycleForBleedingDay } from "./cycle-boundaries";
import { findOwningCycleId } from "./cycle-attribution";
/** The wearables that write period days under this rule. */
export type SubordinateCycleSource = "GARMIN" | "OURA";
/**
* The conservative "logged, intensity unknown" level: a detected bleeding day
* is at least light, never invented as MEDIUM or HEAVY.
*/
export const SUBORDINATE_PERIOD_FLOW: FlowLevel = "LIGHT";
function isUniqueViolation(err: unknown): boolean {
return (
typeof err === "object" &&
err !== null &&
"code" in err &&
(err as { code?: unknown }).code === "P2002"
);
}
/**
* Create the period days no row covers yet, oldest first, each attributed to
* its cycle, and open the cycle a first bleeding day starts. Returns how many
* rows were created.
*/
export async function createSubordinatePeriodDays(
userId: string,
source: SubordinateCycleSource,
days: readonly string[],
): Promise<number> {
if (days.length === 0) return 0;
const user = await prisma.user.findUnique({
where: { id: userId },
select: { timezone: true },
});
const tz = user?.timezone ?? null;
let created = 0;
for (const day of [...new Set(days)].sort()) {
try {
await prisma.cycleDayLog.create({
data: {
userId,
date: day,
tz,
source,
flow: SUBORDINATE_PERIOD_FLOW,
cycleId: await findOwningCycleId(userId, day),
},
});
} catch (err) {
// A row already exists: the person's own, an import's, or this
// source's earlier write. A wearable never overwrites.
if (isUniqueViolation(err)) continue;
getEvent()?.addWarning(
`${source.toLowerCase()}: cycle day-log create failed for ${userId} ${day}: ${err}`,
);
continue;
}
created += 1;
// After the create, so the opened cycle claims the row just written.
await ensureCycleForBleedingDay(userId, day, tz, {
flow: SUBORDINATE_PERIOD_FLOW,
intermenstrualBleeding: false,
});
}
return created;
}
/**
* Stamp the last day of a finished period on the cycle that opens at
* `start`, when that cycle has no period end yet. A person's own end, or one
* an earlier pass stamped, stays. Returns whether a stamp was written.
*/
export async function stampSubordinatePeriodEnd(
userId: string,
start: string,
lastDay: string,
): Promise<boolean> {
if (lastDay < start) return false;
const stamped = await prisma.menstrualCycle.updateMany({
where: {
userId,
startDate: start,
deletedAt: null,
periodEndDate: null,
},
data: { periodEndDate: lastDay, syncVersion: { increment: 1 } },
});
return stamped.count > 0;
}
+17 -6
View File
@@ -370,13 +370,24 @@ describe("cycle summary", () => {
"2026-10-13",
"2026-10-14",
],
ended: true,
});
expect(
periodFromGarminCycleSummary(
{ startDate: "2026-11-03", periodLength: 5 },
"2026-11-04",
)?.days,
).toEqual(["2026-11-03", "2026-11-04"]);
const ongoing = periodFromGarminCycleSummary(
{ startDate: "2026-11-03", periodLength: 5 },
"2026-11-04",
);
expect(ongoing?.days).toEqual(["2026-11-03", "2026-11-04"]);
expect(ongoing?.ended).toBe(false);
});
it("calls a period ended only once its last day is past", () => {
const summary = { startDate: "2026-11-01", periodLength: 4 };
expect(periodFromGarminCycleSummary(summary, "2026-11-04")?.ended).toBe(
false,
);
expect(periodFromGarminCycleSummary(summary, "2026-11-05")?.ended).toBe(
true,
);
});
it("imports nothing from a predicted cycle or a pregnancy", () => {
+28 -47
View File
@@ -4,7 +4,10 @@
* import (`src/lib/oura/cycle-sync.ts`): a day is only ever CREATED when no
* `CycleDayLog` row exists for it at all (any source), and never updated, so
* a person's own entry always wins and an hourly poll can never demote it.
* No `MenstrualCycle` span is created; cycle spans stay an explicit action.
* The write goes through `createSubordinatePeriodDays`, which attributes each
* created day to its cycle and opens the cycle a first bleeding day starts,
* as every other cycle writer does. A finished period also stamps its last
* day on that cycle, when the cycle has no period end yet.
*
* `GET /periodichealth-service/menstrualcycle/summary/{date}` describes the
* cycle a date falls in: its start and its period length. A cycle Garmin
@@ -16,15 +19,15 @@
* Gated behind the cycle module (`isCycleAvailableForUser`), checked before
* Garmin is asked, so an account with cycle tracking off costs no request.
*/
import { prisma } from "@/lib/db";
import { getEvent } from "@/lib/logging/context";
import { isCycleAvailableForUser } from "@/lib/cycle/gate";
import {
createSubordinatePeriodDays,
stampSubordinatePeriodEnd,
} from "@/lib/cycle/subordinate-period-days";
import { GARMIN_PATHS, shiftGarminDay, type GarminSession } from "./client";
import type { GarminCycleDate, GarminCycleSummary } from "./wire-types";
const GARMIN_PERIOD_FLOW = "LIGHT" as const;
/** The longest period a single summary may claim before it is not believed. */
const MAX_PERIOD_DAYS = 14;
@@ -53,6 +56,8 @@ export interface GarminCyclePeriod {
start: string;
/** The period's days, oldest first, none after `today`. */
days: string[];
/** Every day of the period lies before `today`: its last day is known. */
ended: boolean;
}
/** The logged period a summary describes, or null when there is none. */
@@ -67,57 +72,33 @@ export function periodFromGarminCycleSummary(
const length = summary.periodLength;
if (typeof length !== "number" || !Number.isFinite(length)) return null;
const n = Math.round(length);
if (n < 1 || n > MAX_PERIOD_DAYS) return { start, days: [] };
if (n < 1 || n > MAX_PERIOD_DAYS) return { start, days: [], ended: false };
const days: string[] = [];
for (let i = 0; i < n; i++) {
const day = shiftGarminDay(start, i);
if (day > today) break;
days.push(day);
}
return { start, days };
const ended = days.length === n && days[n - 1]! < today;
return { start, days, ended };
}
function isUniqueViolation(err: unknown): boolean {
return (
typeof err === "object" &&
err !== null &&
"code" in err &&
(err as { code?: unknown }).code === "P2002"
);
}
/** Create the period days no row covers yet. Returns how many were created. */
export async function writeGarminPeriodDays(
/**
* Write one period: create the days no row covers yet (opening and
* attributing its cycle), then stamp the period's end once it is over.
* Returns how many days were created.
*/
export async function writeGarminPeriod(
userId: string,
days: readonly string[],
period: GarminCyclePeriod,
): Promise<number> {
if (days.length === 0) return 0;
const user = await prisma.user.findUnique({
where: { id: userId },
select: { timezone: true },
});
const tz = user?.timezone ?? null;
let created = 0;
for (const day of days) {
try {
await prisma.cycleDayLog.create({
data: {
userId,
date: day,
tz,
source: "GARMIN",
flow: GARMIN_PERIOD_FLOW,
},
});
created += 1;
} catch (err) {
// A row already exists: the person's own, an import's, or this
// module's earlier write. Garmin never overwrites.
if (isUniqueViolation(err)) continue;
getEvent()?.addWarning(
`garmin: cycle day-log create failed for ${userId} ${day}: ${err}`,
);
}
const created = await createSubordinatePeriodDays(
userId,
"GARMIN",
period.days,
);
if (period.ended) {
await stampSubordinatePeriodEnd(userId, period.start, period.days.at(-1)!);
}
return created;
}
@@ -147,5 +128,5 @@ export async function syncGarminCycle(
const summary = await readGarminCycleSummary(session, today);
const period = periodFromGarminCycleSummary(summary, today);
if (!period) return 0;
return writeGarminPeriodDays(userId, period.days);
return writeGarminPeriod(userId, period);
}
@@ -114,12 +114,12 @@ describe("maintenance group queue policies", () => {
// One-shot repairs on a fixed singleton key: a boot during a run must be
// able to queue the next pass, so only queued duplicates collapse.
it.each([["lab-document-date-repair", "short"]])(
"%s is %s",
(queue, policy) => {
expectPolicy("maintenance", queue, policy);
},
);
it.each([
["lab-document-date-repair", "short"],
["wearable-cycle-repair", "short"],
])("%s is %s", (queue, policy) => {
expectPolicy("maintenance", queue, policy);
});
it("environment-fetch is deliberately left on standard", () => {
// Its explicit-range backfill sends with no key at all, on purpose, so any
+2 -2
View File
@@ -44,7 +44,7 @@ import {
import {
periodFromGarminCycleSummary,
readGarminCycleSummary,
writeGarminPeriodDays,
writeGarminPeriod,
} from "@/lib/garmin/cycle-sync";
import { garminUnavailableCode } from "@/lib/garmin/unavailable";
import {
@@ -212,7 +212,7 @@ async function readStepWindow(
const period = periodFromGarminCycleSummary(summary, progress.anchor);
// No logged cycle at or before this day: the walk is over.
if (!period || period.start > state.before) return null;
await writeGarminPeriodDays(userId, period.days);
await writeGarminPeriod(userId, period);
state.before = shiftDateKey(period.start, -1);
return 0;
}
+3
View File
@@ -493,6 +493,9 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
"lab-document-date-repair": short(
"Reads the document lab values still filed at their confirmation moment, a handful per account, and moves each with one update.",
),
"wearable-cycle-repair": short(
"Reads the wearable period days still without a cycle, a few hundred per account at most, and opens or attributes each with a short transaction.",
),
"measurement-maintenance": {
runtime: "long",
expireInSeconds: MEASUREMENT_MAINTENANCE_SEND_OPTIONS.expireInSeconds,
@@ -86,6 +86,15 @@ export const MAINTENANCE_DISCOVERY: readonly BootDiscovery[] = [
await import("@/lib/jobs/lab-document-date-repair")
).enqueueBootTimeLabDocumentDateRepair(),
},
// Converges to nothing once the wearable period days have their cycles; a
// boot with none costs one query.
{
name: "wearable-cycle-repair",
run: async () =>
(
await import("@/lib/jobs/wearable-cycle-repair")
).enqueueBootTimeWearableCycleRepair(),
},
// Self-limiting and idempotent; a boot with no backlog costs one query.
{
name: "compaction-tombstone-purge",
@@ -5,6 +5,7 @@
* something to repair.
*/
import { LAB_DOCUMENT_DATE_REPAIR_QUEUE } from "@/lib/jobs/lab-document-date-repair-queue";
import { WEARABLE_CYCLE_REPAIR_QUEUE } from "@/lib/jobs/wearable-cycle-repair-queue";
import { defineJob, type JobDefinition } from "./define-job";
@@ -22,4 +23,17 @@ export const MAINTENANCE_REPAIR_JOBS: readonly JobDefinition[] = [
(await import("@/lib/jobs/lab-document-date-repair"))
.handleLabDocumentDateRepair,
}),
// Wearable period days imported without the cycles they open.
defineJob({
queue: WEARABLE_CYCLE_REPAIR_QUEUE,
policy: {
policy: "short",
reason:
"Fixed singleton key; collapse queued duplicates only, so a boot during a run can queue the next one.",
},
work: { localConcurrency: 1 },
load: async () =>
(await import("@/lib/jobs/wearable-cycle-repair"))
.handleWearableCycleRepair,
}),
];
+1
View File
@@ -124,6 +124,7 @@ const subsystemSurface = {
"compaction-tombstone-purge": { audience: "system" },
"measurement-fold-repair": { audience: "system" },
"lab-document-date-repair": { audience: "system" },
"wearable-cycle-repair": { audience: "system" },
"measurement-maintenance": { audience: "system" },
"pulse-day-mean": { audience: "system" },
"medication-intake-import": { audience: "system" },
@@ -0,0 +1,9 @@
/**
* The queue facts of `wearable-cycle-repair.ts`: its name and its one
* singleton key. They live apart from the job so the worker registry and the
* boot discovery can read them without loading it.
*/
export const WEARABLE_CYCLE_REPAIR_QUEUE = "wearable-cycle-repair";
/** One pass covers every account, so there is only ever one job. */
export const WEARABLE_CYCLE_REPAIR_SINGLETON_KEY = "wearable-cycle-repair";
+271
View File
@@ -0,0 +1,271 @@
/**
* One-shot repair: period days Garmin or Oura imported before their write
* learned to open and attribute cycles.
*
* Both wearables wrote their period days as bare rows, past the seam every
* other cycle writer goes through (`ensureCycleForBleedingDay`), so an account
* whose periods came from a watch or a ring held day-logs and no cycle: no
* start, no lengths, no forecast. The write now goes through
* `createSubordinatePeriodDays`; this pass gives the rows written before it
* the cycles they should have opened, by the same rule, oldest first per
* account.
*
* A row is a candidate when it is a live GARMIN or OURA bleeding day without
* a cycle and the rule can still place it: either a cycle starts on or before
* it (the row only needs its attribution), or no cycle does and nothing in the
* way stops it opening one (`ensureCycleForBleedingDay` declines a day that
* lies too close before a later start without being the same period). A day
* the rule declines stays unattributed and drops out of the predicate, so the
* pass converges: a second run finds nothing, and a boot with nothing to
* repair costs one query.
*
* Only cycles and the `cycleId` link move. No day-log's own content is
* touched, a person's own rows included. A Garmin period whose cycle this pass
* opened and that is over (a later cycle has started) has its last day stamped
* as the period end, from the run of consecutive Garmin days at its start.
*/
import type { Job } from "pg-boss";
import { Prisma, type PrismaClient } from "@/generated/prisma/client";
import { activeAccountSql } from "@/lib/account-deletion/active-account";
import { ensureCycleForBleedingDay } from "@/lib/cycle/cycle-boundaries";
import { addDays } from "@/lib/cycle/day-math";
import { stampSubordinatePeriodEnd } from "@/lib/cycle/subordinate-period-days";
import { HARD_CYCLE_MIN, PERIOD_MAX } from "@/lib/cycle/types";
import { prisma as defaultPrisma } from "@/lib/db";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
import { workerLog } from "@/lib/jobs/reminder/shared";
import { annotate } from "@/lib/logging/context";
import { withBackgroundEvent } from "@/lib/logging/background";
import {
WEARABLE_CYCLE_REPAIR_QUEUE,
WEARABLE_CYCLE_REPAIR_SINGLETON_KEY,
} from "./wearable-cycle-repair-queue";
/** The longest run of Garmin days read as one period. */
const MAX_PERIOD_RUN_DAYS = 14;
interface CandidateRow {
id: string;
userId: string;
date: string;
}
/**
* The SQL half of the rule. `limit` turns it into the boot's existence check.
* The second arm mirrors `ensureCycleForBleedingDay`: with no cycle on or
* before the day, it opens one unless a later start lies within the hard
* minimum cycle length without being the same period.
*/
function candidateQuery(limit: number | null): Prisma.Sql {
return Prisma.sql`
SELECT d."id", d."user_id" AS "userId", d."date"
FROM "cycle_day_logs" d
WHERE d."source" IN ('GARMIN', 'OURA')
AND d."deleted_at" IS NULL
AND d."cycle_id" IS NULL
AND d."flow" IS NOT NULL
AND d."flow" NOT IN ('NONE', 'SPOTTING')
AND NOT d."intermenstrual_bleeding"
AND (
EXISTS (
SELECT 1 FROM "menstrual_cycles" c
WHERE c."user_id" = d."user_id" AND c."deleted_at" IS NULL
AND c."start_date" <= d."date"
)
OR NOT EXISTS (
SELECT 1 FROM "menstrual_cycles" c
WHERE c."user_id" = d."user_id" AND c."deleted_at" IS NULL
AND c."start_date" >= to_char(d."date"::date + ${PERIOD_MAX}::int, 'YYYY-MM-DD')
AND c."start_date" < to_char(d."date"::date + ${HARD_CYCLE_MIN}::int, 'YYYY-MM-DD')
)
)
AND ${activeAccountSql('d."user_id"')}
ORDER BY d."user_id", d."date"
${limit === null ? Prisma.empty : Prisma.sql`LIMIT ${limit}`}
`;
}
export interface WearableCycleRepairResult {
/** Rows the SQL predicate returned. */
scanned: number;
/** Cycles a row opened. */
opened: number;
/** Rows given the cycle they fall in. */
attributed: number;
/** Garmin period ends stamped on an opened cycle. */
periodEnds: number;
/** Accounts touched. */
accounts: number;
}
/** The last day of the run of consecutive Garmin days starting at `start`. */
async function garminRunEnd(
prisma: PrismaClient,
userId: string,
start: string,
): Promise<string | null> {
const rows = await prisma.cycleDayLog.findMany({
where: {
userId,
source: "GARMIN",
deletedAt: null,
date: { gte: start, lt: addDays(start, MAX_PERIOD_RUN_DAYS) },
},
orderBy: { date: "asc" },
select: { date: true },
});
let end: string | null = null;
for (const row of rows) {
if (row.date !== (end === null ? start : addDays(end, 1))) break;
end = row.date;
}
return end;
}
/** Repair one account's rows, oldest first. */
async function repairAccount(
prisma: PrismaClient,
userId: string,
rows: readonly CandidateRow[],
result: WearableCycleRepairResult,
): Promise<void> {
const user = await prisma.user.findUnique({
where: { id: userId },
select: { timezone: true },
});
const tz = user?.timezone ?? null;
const openedGarminStarts: string[] = [];
for (const row of rows) {
// Re-read. A cycle opened for an earlier row claims every day after it
// until the next start, this one included, so a claimed row is still
// asked whether it starts a cycle of its own.
const live = await prisma.cycleDayLog.findFirst({
where: { id: row.id, deletedAt: null },
select: {
source: true,
flow: true,
intermenstrualBleeding: true,
cycleId: true,
},
});
if (!live) continue;
const opened = await ensureCycleForBleedingDay(userId, row.date, tz, {
flow: live.flow,
intermenstrualBleeding: live.intermenstrualBleeding,
});
if (opened) {
result.opened += 1;
if (live.source === "GARMIN") openedGarminStarts.push(row.date);
continue;
}
if (live.cycleId !== null) continue;
const owning = await prisma.menstrualCycle.findFirst({
where: { userId, deletedAt: null, startDate: { lte: row.date } },
orderBy: { startDate: "desc" },
select: { id: true },
});
if (!owning) continue;
const moved = await prisma.cycleDayLog.updateMany({
where: { id: row.id, cycleId: null },
data: { cycleId: owning.id, syncVersion: { increment: 1 } },
});
result.attributed += moved.count;
}
for (const start of openedGarminStarts) {
const cycle = await prisma.menstrualCycle.findFirst({
where: { userId, startDate: start, deletedAt: null },
select: { endDate: true },
});
// An open cycle may still be bleeding; the hourly sync stamps its end.
if (!cycle?.endDate) continue;
const end = await garminRunEnd(prisma, userId, start);
if (end && (await stampSubordinatePeriodEnd(userId, start, end))) {
result.periodEnds += 1;
}
}
}
/** The repair itself, across every account. */
export async function repairWearableCycleDays(
prisma: PrismaClient = defaultPrisma,
): Promise<WearableCycleRepairResult> {
const rows = await prisma.$queryRaw<CandidateRow[]>(candidateQuery(null));
const byUser = new Map<string, CandidateRow[]>();
for (const row of rows) {
const list = byUser.get(row.userId) ?? [];
list.push(row);
byUser.set(row.userId, list);
}
const result: WearableCycleRepairResult = {
scanned: rows.length,
opened: 0,
attributed: 0,
periodEnds: 0,
accounts: byUser.size,
};
for (const [userId, list] of byUser) {
await repairAccount(prisma, userId, list, result);
}
return result;
}
/** The queue handler: one run over every account. */
export async function handleWearableCycleRepair(
_jobs: Job<Record<string, never>>[],
prisma: PrismaClient = defaultPrisma,
): Promise<JobOutcome> {
return withBackgroundEvent("job.wearable_cycle_repair", async () => {
let result: WearableCycleRepairResult;
try {
result = await repairWearableCycleDays(prisma);
} catch (err) {
workerLog("error", "[wearable-cycle-repair] failed", err);
return jobFailed("wearable cycle repair failed", err);
}
workerLog(
"info",
`[wearable-cycle-repair] scanned=${result.scanned} opened=${result.opened} attributed=${result.attributed} periodEnds=${result.periodEnds} accounts=${result.accounts}`,
);
annotate({
action: { name: "cycle.wearableDays.repair" },
meta: { ...result },
});
return jobDone({
candidates_scanned: result.scanned,
// Cycles opened, rows attributed, period ends stamped, accounts.
created: result.opened,
linked: result.attributed,
updated: result.periodEnds,
users: result.accounts,
});
});
}
/**
* Boot discovery: one existence query, and the one job only when there is
* something to repair.
*/
export async function enqueueBootTimeWearableCycleRepair(
prisma: PrismaClient = defaultPrisma,
): Promise<{ enqueued: boolean; error: string | null }> {
const boss = getGlobalBoss();
if (!boss) return { enqueued: false, error: null };
try {
const any = await prisma.$queryRaw<CandidateRow[]>(candidateQuery(1));
if (any.length === 0) return { enqueued: false, error: null };
const id = await boss.send(
WEARABLE_CYCLE_REPAIR_QUEUE,
{},
{ singletonKey: WEARABLE_CYCLE_REPAIR_SINGLETON_KEY, retryLimit: 2 },
);
return { enqueued: id !== null, error: null };
} catch (err) {
return {
enqueued: false,
error: err instanceof Error ? err.message : String(err),
};
}
}
+56 -7
View File
@@ -6,13 +6,21 @@
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
const { fetchCyclePhasesMock, isCycleAvailableMock, findUserMock, createMock } =
vi.hoisted(() => ({
fetchCyclePhasesMock: vi.fn(),
isCycleAvailableMock: vi.fn(),
findUserMock: vi.fn(),
createMock: vi.fn(),
}));
const {
fetchCyclePhasesMock,
isCycleAvailableMock,
findUserMock,
createMock,
ensureCycleMock,
owningCycleMock,
} = vi.hoisted(() => ({
fetchCyclePhasesMock: vi.fn(),
isCycleAvailableMock: vi.fn(),
findUserMock: vi.fn(),
createMock: vi.fn(),
ensureCycleMock: vi.fn(),
owningCycleMock: vi.fn(),
}));
vi.mock("@/lib/db", () => ({
prisma: {
@@ -21,6 +29,14 @@ vi.mock("@/lib/db", () => ({
},
}));
vi.mock("@/lib/cycle/cycle-boundaries", () => ({
ensureCycleForBleedingDay: ensureCycleMock,
}));
vi.mock("@/lib/cycle/cycle-attribution", () => ({
findOwningCycleId: owningCycleMock,
}));
vi.mock("@/lib/cycle/gate", () => ({
isCycleAvailableForUser: isCycleAvailableMock,
}));
@@ -50,6 +66,8 @@ beforeEach(() => {
isCycleAvailableMock.mockReset().mockResolvedValue(true);
findUserMock.mockReset().mockResolvedValue({ timezone: "Europe/Berlin" });
createMock.mockReset().mockResolvedValue({ id: "cdl-1" });
ensureCycleMock.mockReset().mockResolvedValue(null);
owningCycleMock.mockReset().mockResolvedValue(null);
});
describe("syncUserOuraCyclePhases", () => {
@@ -93,10 +111,41 @@ describe("syncUserOuraCyclePhases", () => {
tz: "Europe/Berlin",
source: "OURA",
flow: "LIGHT",
cycleId: null,
},
});
});
it("attributes a created day to its cycle and lets a first bleeding day open one", async () => {
fetchCyclePhasesMock.mockResolvedValue([
{ day: "2026-06-05", phase: "menstrual" },
]);
owningCycleMock.mockResolvedValue("cycle-before");
await syncUserOuraCyclePhases("u1", "tok", WINDOW);
expect(createMock).toHaveBeenCalledWith({
data: expect.objectContaining({ cycleId: "cycle-before" }),
});
expect(ensureCycleMock).toHaveBeenCalledWith(
"u1",
"2026-06-05",
"Europe/Berlin",
{ flow: "LIGHT", intermenstrualBleeding: false },
);
// The inference runs after the create, so the opened cycle claims it.
expect(createMock.mock.invocationCallOrder[0]).toBeLessThan(
ensureCycleMock.mock.invocationCallOrder[0]!,
);
});
it("infers no cycle from a day it did not create", async () => {
fetchCyclePhasesMock.mockResolvedValue([
{ day: "2026-06-05", phase: "menstrual" },
]);
createMock.mockRejectedValue(uniqueViolation());
await syncUserOuraCyclePhases("u1", "tok", WINDOW);
expect(ensureCycleMock).not.toHaveBeenCalled();
});
it("skips a day that already has ANY existing row (manual wins) — unique-violation swallowed silently", async () => {
fetchCyclePhasesMock.mockResolvedValue([
{ day: "2026-06-05", phase: "menstrual" },
+12 -62
View File
@@ -17,45 +17,26 @@
* So this module never touches the shared upsert helper. It only ever
* CREATEs a day that has NO existing `CycleDayLog` row at all (any source),
* and never updates a row it did not itself create. A day already owned by
* this module (source = OURA) is left alone on re-sync too — the marker is
* a fixed single-value hint (`flow: LIGHT`), not a corrigible reading, so
* there is nothing to refresh. The whole path degenerates to a cheap
* "create if absent, otherwise no-op".
* this module (source = OURA) is left alone on re-sync too: the marker is a
* fixed single-value hint (`flow: LIGHT`), not a corrigible reading, so there
* is nothing to refresh.
*
* The create goes through `createSubordinatePeriodDays`, which does what the
* shared upsert does around the row: the day is attributed to the cycle it
* falls in, and a first bleeding day opens its cycle, the same inference the
* single capture, the bulk drain, the Apple Health fold and the Health Connect
* import run. Oura reports no period length, so no period end is stamped.
*
* Gated behind the SAME two-layer cycle-module gate every other cycle
* writer respects (`isCycleAvailableForUser`) so an operator-disabled
* instance, or a user who turned cycle tracking off, never accumulates rows
* from a background Oura poll. No `MenstrualCycle` span is ever created or
* mutated here — mirrors the Apple Health importer precedent
* (`src/lib/cycle/import-accumulator.ts`), which attributes day-logs to an
* EXISTING cycle but never synthesises a new one from flow data either;
* cycle-span creation stays an explicit user action
* (`POST /api/cycle/period`).
* from a background Oura poll.
*/
import { prisma } from "@/lib/db";
import { getEvent } from "@/lib/logging/context";
import { isCycleAvailableForUser } from "@/lib/cycle/gate";
import { createSubordinatePeriodDays } from "@/lib/cycle/subordinate-period-days";
import { fetchDailyCyclePhases } from "./client";
import { derivePeriodDaysFromCyclePhases } from "./map";
/** The conservative "logged but unspecified intensity" flow level — mirrors
* the HealthKit `unspecified` → `LIGHT` boundary documented in
* `healthkit-mapping.ts`: a detected-but-unmeasured bleeding day is at least
* light flow, never invented as MEDIUM/HEAVY. */
const OURA_PERIOD_FLOW = "LIGHT" as const;
/** Prisma P2002 (unique constraint) detector, mirrored locally from
* `day-log-write.ts` rather than imported — a two-line duplicate is cheaper
* than coupling this module to that helper's internals. */
function isUniqueViolation(err: unknown): boolean {
return (
typeof err === "object" &&
err !== null &&
"code" in err &&
(err as { code?: unknown }).code === "P2002"
);
}
/**
* Sync one user's Oura `daily_cycle_phases` window into `CycleDayLog`.
* Returns the count of days newly created.
@@ -88,36 +69,5 @@ export async function syncUserOuraCyclePhases(
const available = await isCycleAvailableForUser(userId);
if (!available) return 0;
const user = await prisma.user.findUnique({
where: { id: userId },
select: { timezone: true },
});
const tz = user?.timezone ?? null;
let created = 0;
for (const day of periodDays) {
try {
await prisma.cycleDayLog.create({
data: {
userId,
date: day,
tz,
source: "OURA",
flow: OURA_PERIOD_FLOW,
},
});
created += 1;
} catch (err) {
if (isUniqueViolation(err)) {
// A row already exists for this date — the user's own entry, an
// Apple Health import, or this module's own prior write. Never
// overwrite: Oura is a subordinate signal here.
continue;
}
getEvent()?.addWarning(
`oura: cycle day-log create failed for ${userId} ${day}: ${err}`,
);
}
}
return created;
return createSubordinatePeriodDays(userId, "OURA", periodDays);
}
@@ -1,5 +1,5 @@
{
"registeredQueueCount": 129,
"registeredQueueCount": 130,
"registeredQueues": [
"account-deletion-purge",
"achievement-unlock-sweep",
@@ -117,6 +117,7 @@
"supplement-nutrient-recompute",
"system-job-alert",
"tls-pin-monitor",
"wearable-cycle-repair",
"whoop-backfill",
"whoop-cycle-sync",
"whoop-oauth-state-cleanup",
@@ -1872,6 +1873,21 @@
"retry_delay_max": null,
"retry_limit": 2
},
{
"dead_letter": null,
"deletion_seconds": 604800,
"expire_seconds": 900,
"heartbeat_seconds": null,
"name": "wearable-cycle-repair",
"notify": false,
"partition": false,
"policy": "short",
"retention_seconds": 1209600,
"retry_backoff": false,
"retry_delay": 0,
"retry_delay_max": null,
"retry_limit": 2
},
{
"dead_letter": null,
"deletion_seconds": 604800,
@@ -3801,6 +3817,12 @@
"pollingIntervalSeconds": 30
}
},
{
"queue": "wearable-cycle-repair",
"options": {
"localConcurrency": 1
}
},
{
"queue": "whoop-backfill",
"options": {
+131
View File
@@ -77,6 +77,8 @@ const fake = vi.hoisted(() => ({
rateLimitHit: -1,
/** First day of the one logged cycle, relative to today. */
cycleStart: "",
/** Several logged cycles instead of the one; when set, `cycleStart` is unused. */
cycles: [] as Array<{ start: string; periodLength: number }>,
/**
* The weigh-in shape: `spec` is Garmin's document (kilograms, no
* `samplePk`, a string instant), `live` the shape seen on the live
@@ -260,6 +262,18 @@ function answer(path: string, q: URLSearchParams): unknown {
if (path.startsWith("/periodichealth-service/menstrualcycle/summary/")) {
if (fake.cycleStatus !== 200) return { status: fake.cycleStatus };
const day = path.split("/").pop()!;
if (fake.cycles.length > 0) {
const at = [...fake.cycles]
.sort((a, b) => b.start.localeCompare(a.start))
.find((c) => c.start <= day);
return at
? {
startDate: { localDate: at.start },
periodLength: at.periodLength,
predictedCycle: false,
}
: {};
}
const cycleStart = fake.cycleStart;
return day >= cycleStart
? { startDate: cycleStart, periodLength: 4, predictedCycle: false }
@@ -362,6 +376,7 @@ beforeEach(async () => {
fake.today = new Date().toISOString().slice(0, 10);
fake.historyFrom = addDays(fake.today, -20);
fake.cycleStart = addDays(fake.today, -15);
fake.cycles = [];
fake.weightShape = "spec";
fake.extraWeighDays = [];
fake.cycleStatus = 200;
@@ -596,6 +611,122 @@ describe("Garmin history backfill", () => {
});
});
describe("Garmin cycle history", () => {
async function backfillToEnd() {
const { runGarminBackfillForUser } =
await import("@/lib/jobs/garmin-backfill");
for (let slice = 0; slice < 20; slice++) {
const user = await getPrismaClient().user.findUnique({
where: { id: USER_ID },
select: { garminBackfillCompletedAt: true },
});
if (user?.garminBackfillCompletedAt) return;
await runGarminBackfillForUser(USER_ID, () => false, slice);
}
throw new Error("backfill did not finish");
}
async function liveCycles() {
return getPrismaClient().menstrualCycle.findMany({
where: { userId: USER_ID, deletedAt: null },
orderBy: { startDate: "asc" },
select: {
id: true,
startDate: true,
endDate: true,
lengthDays: true,
periodEndDate: true,
},
});
}
it("opens a cycle per back-filled period, newest first, with lengths and a forecast", async () => {
await connectUser();
fake.historyFrom = addDays(fake.today, -100);
const starts = [-80, -52, -24].map((n) => addDays(fake.today, n));
fake.cycles = starts.map((start) => ({ start, periodLength: 5 }));
// A day the person logged inside the middle period.
const ownDay = addDays(starts[1]!, 2);
const prisma = getPrismaClient();
await prisma.cycleDayLog.create({
data: { userId: USER_ID, date: ownDay, flow: "HEAVY" },
});
await backfillToEnd();
const cycles = await liveCycles();
expect(
cycles.map((c) => [
c.startDate,
c.endDate,
c.lengthDays,
c.periodEndDate,
]),
).toEqual([
[starts[0], addDays(starts[1]!, -1), 28, addDays(starts[0]!, 4)],
[starts[1], addDays(starts[2]!, -1), 28, addDays(starts[1]!, 4)],
[starts[2], null, null, addDays(starts[2]!, 4)],
]);
// Every Garmin day belongs to the period it is part of.
const days = await prisma.cycleDayLog.findMany({
where: { userId: USER_ID },
orderBy: { date: "asc" },
select: { date: true, source: true, flow: true, cycleId: true },
});
expect(days).toHaveLength(15);
for (const day of days) {
const owner = [...cycles].reverse().find((c) => c.startDate <= day.date);
expect(day.cycleId).toBe(owner?.id);
}
// The person's own day is theirs, as they logged it.
expect(days.find((d) => d.date === ownDay)).toMatchObject({
source: "MANUAL",
flow: "HEAVY",
});
const { getOrCreateCycleProfile } = await import("@/lib/cycle/profile");
await getOrCreateCycleProfile(USER_ID);
const { refreshPredictionCacheForUser } =
await import("@/lib/cycle/prediction-refresh");
expect(await refreshPredictionCacheForUser(USER_ID)).toBe("persisted");
const forecast = await prisma.cyclePrediction.findUnique({
where: { userId: USER_ID },
});
expect(forecast).not.toBeNull();
// A second full backfill changes nothing.
await prisma.user.update({
where: { id: USER_ID },
data: {
garminBackfillProgress: Prisma.DbNull,
garminBackfillCompletedAt: null,
},
});
await backfillToEnd();
expect(await liveCycles()).toEqual(cycles);
expect(await prisma.cycleDayLog.count({ where: { userId: USER_ID } })).toBe(
15,
);
});
it("opens the current cycle from the hourly sync and leaves its end open while bleeding", async () => {
await connectUser();
const start = addDays(fake.today, -1);
fake.cycles = [{ start, periodLength: 5 }];
const { syncUserGarmin } = await import("@/lib/garmin/sync");
await syncUserGarmin(USER_ID, { lookbackDays: 2 });
const cycles = await liveCycles();
expect(cycles).toHaveLength(1);
expect(cycles[0]).toMatchObject({ startDate: start, periodEndDate: null });
const days = await getPrismaClient().cycleDayLog.findMany({
where: { userId: USER_ID },
select: { cycleId: true },
});
expect(days).toHaveLength(2);
expect(days.every((d) => d.cycleId === cycles[0]!.id)).toBe(true);
});
});
describe("Garmin connect and disconnect", () => {
function post(url: string, body?: unknown) {
return new NextRequest(`http://localhost${url}`, {
@@ -0,0 +1,200 @@
/**
* The one-shot repair of wearable period days imported without the cycles
* they open (real Postgres).
*
* Garmin and Oura used to write their period days as bare rows, so an
* account whose periods came from a watch or a ring held day-logs and no
* cycle. The repair opens the cycles those rows start, attributes the rest,
* stamps the end of a finished Garmin period, and touches no day-log's own
* content. A day the cycle rule declines stays as it is and drops out of the
* candidate set, so the boot check converges; a second run changes nothing.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
import { getPrismaClient, truncateAllTables } from "./setup";
const sends = vi.hoisted(() => [] as string[]);
vi.mock("@/lib/jobs/boss-instance", () => ({
getGlobalBoss: () => ({
send: async (queue: string) => {
sends.push(queue);
return `job-${sends.length}`;
},
}),
}));
import {
enqueueBootTimeWearableCycleRepair,
repairWearableCycleDays,
} from "@/lib/jobs/wearable-cycle-repair";
const GARMIN_USER = "user-wearable-repair-garmin";
const OURA_USER = "user-wearable-repair-oura";
const BLOCKED_USER = "user-wearable-repair-blocked";
const DAY_MS = 86_400_000;
const addDays = (day: string, n: number) =>
new Date(Date.parse(`${day}T00:00:00Z`) + n * DAY_MS)
.toISOString()
.slice(0, 10);
const GARMIN_STARTS = ["2026-07-01", "2026-07-29", "2026-08-26"];
async function seedUser(id: string) {
await getPrismaClient().user.create({
data: {
id,
username: id,
email: `${id}@example.test`,
timezone: "UTC",
gender: "FEMALE",
},
});
}
/** Day-logs as a v1.47.0 import left them: no cycle. */
async function orphanDays(
userId: string,
source: "GARMIN" | "OURA",
days: string[],
) {
await getPrismaClient().cycleDayLog.createMany({
data: days.map((date) => ({
userId,
date,
tz: "UTC",
source,
flow: "LIGHT" as const,
})),
});
}
async function liveCycles(userId: string) {
return getPrismaClient().menstrualCycle.findMany({
where: { userId, deletedAt: null },
orderBy: { startDate: "asc" },
select: {
id: true,
startDate: true,
endDate: true,
lengthDays: true,
periodEndDate: true,
syncVersion: true,
},
});
}
beforeEach(async () => {
const prisma = getPrismaClient();
await truncateAllTables(prisma);
sends.length = 0;
for (const id of [GARMIN_USER, OURA_USER, BLOCKED_USER]) await seedUser(id);
await orphanDays(
GARMIN_USER,
"GARMIN",
GARMIN_STARTS.flatMap((start) =>
[0, 1, 2, 3, 4].map((i) => addDays(start, i)),
).filter((d) => d !== "2026-07-31"),
);
// The person's own day inside the middle period, logged before any cycle.
await prisma.cycleDayLog.create({
data: {
userId: GARMIN_USER,
date: "2026-07-31",
flow: "HEAVY",
source: "MANUAL",
},
});
await orphanDays(OURA_USER, "OURA", ["2026-06-01", "2026-06-29"]);
// A Garmin day 15 days before a start the person tapped: too close to be a
// cycle of its own, too far to be the same period. The rule declines it.
await prisma.menstrualCycle.create({
data: { userId: BLOCKED_USER, startDate: "2026-09-16", tz: "UTC" },
});
await orphanDays(BLOCKED_USER, "GARMIN", ["2026-09-01"]);
});
describe("wearable cycle repair", () => {
it("gives orphaned wearable period days their cycles", async () => {
expect(await enqueueBootTimeWearableCycleRepair()).toEqual({
enqueued: true,
error: null,
});
const result = await repairWearableCycleDays();
expect(result).toMatchObject({ opened: 5, periodEnds: 2, accounts: 2 });
const garmin = await liveCycles(GARMIN_USER);
expect(
garmin.map((c) => [
c.startDate,
c.endDate,
c.lengthDays,
c.periodEndDate,
]),
).toEqual([
["2026-07-01", "2026-07-28", 28, "2026-07-05"],
// The person's day breaks the Garmin run, so the end is the day before.
["2026-07-29", "2026-08-25", 28, "2026-07-30"],
// Still open: the hourly sync stamps its end.
["2026-08-26", null, null, null],
]);
const prisma = getPrismaClient();
const days = await prisma.cycleDayLog.findMany({
where: { userId: GARMIN_USER },
orderBy: { date: "asc" },
});
for (const day of days) {
const owner = [...garmin].reverse().find((c) => c.startDate <= day.date);
expect(day.cycleId).toBe(owner!.id);
}
expect(days.find((d) => d.date === "2026-07-31")).toMatchObject({
source: "MANUAL",
flow: "HEAVY",
});
const oura = await liveCycles(OURA_USER);
expect(oura.map((c) => [c.startDate, c.lengthDays])).toEqual([
["2026-06-01", 28],
["2026-06-29", null],
]);
// The declined day stays as it was and no longer counts as work.
const blocked = await prisma.cycleDayLog.findFirst({
where: { userId: BLOCKED_USER },
});
expect(blocked?.cycleId).toBeNull();
expect(await liveCycles(BLOCKED_USER)).toHaveLength(1);
sends.length = 0;
expect(await enqueueBootTimeWearableCycleRepair()).toEqual({
enqueued: false,
error: null,
});
expect(sends).toEqual([]);
});
it("changes nothing on a second run", async () => {
await repairWearableCycleDays();
const before = await liveCycles(GARMIN_USER);
const daysBefore = await getPrismaClient().cycleDayLog.findMany({
where: { userId: GARMIN_USER },
orderBy: { date: "asc" },
select: { id: true, cycleId: true, syncVersion: true, flow: true },
});
const again = await repairWearableCycleDays();
expect(again).toMatchObject({ scanned: 0, opened: 0, attributed: 0 });
expect(await liveCycles(GARMIN_USER)).toEqual(before);
expect(
await getPrismaClient().cycleDayLog.findMany({
where: { userId: GARMIN_USER },
orderBy: { date: "asc" },
select: { id: true, cycleId: true, syncVersion: true, flow: true },
}),
).toEqual(daysBefore);
});
});