diff --git a/CHANGELOG.md b/CHANGELOG.md index 97946122c..45d96f795 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/src/lib/cycle/subordinate-period-days.ts b/src/lib/cycle/subordinate-period-days.ts new file mode 100644 index 000000000..569451520 --- /dev/null +++ b/src/lib/cycle/subordinate-period-days.ts @@ -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 { + 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 { + 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; +} diff --git a/src/lib/garmin/__tests__/map.test.ts b/src/lib/garmin/__tests__/map.test.ts index 6a01d2e10..83050b5ee 100644 --- a/src/lib/garmin/__tests__/map.test.ts +++ b/src/lib/garmin/__tests__/map.test.ts @@ -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", () => { diff --git a/src/lib/garmin/cycle-sync.ts b/src/lib/garmin/cycle-sync.ts index b76c78f21..a826847e0 100644 --- a/src/lib/garmin/cycle-sync.ts +++ b/src/lib/garmin/cycle-sync.ts @@ -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 { - 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); } diff --git a/src/lib/jobs/__tests__/queue-policy-tables.test.ts b/src/lib/jobs/__tests__/queue-policy-tables.test.ts index 41c5b2f3d..b3e6004b6 100644 --- a/src/lib/jobs/__tests__/queue-policy-tables.test.ts +++ b/src/lib/jobs/__tests__/queue-policy-tables.test.ts @@ -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 diff --git a/src/lib/jobs/garmin-backfill.ts b/src/lib/jobs/garmin-backfill.ts index 7409a03d8..c9b37b029 100644 --- a/src/lib/jobs/garmin-backfill.ts +++ b/src/lib/jobs/garmin-backfill.ts @@ -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; } diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 94b0107d0..3f65887bc 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -493,6 +493,9 @@ export const QUEUE_RUNTIME: Readonly> = { "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, diff --git a/src/lib/jobs/registry/maintenance-discovery.ts b/src/lib/jobs/registry/maintenance-discovery.ts index 2de44262a..d4dbbe866 100644 --- a/src/lib/jobs/registry/maintenance-discovery.ts +++ b/src/lib/jobs/registry/maintenance-discovery.ts @@ -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", diff --git a/src/lib/jobs/registry/maintenance-repairs.ts b/src/lib/jobs/registry/maintenance-repairs.ts index 2d3224e86..d76cf7b14 100644 --- a/src/lib/jobs/registry/maintenance-repairs.ts +++ b/src/lib/jobs/registry/maintenance-repairs.ts @@ -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, + }), ]; diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index b94351e76..5296c702d 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -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" }, diff --git a/src/lib/jobs/wearable-cycle-repair-queue.ts b/src/lib/jobs/wearable-cycle-repair-queue.ts new file mode 100644 index 000000000..d106718aa --- /dev/null +++ b/src/lib/jobs/wearable-cycle-repair-queue.ts @@ -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"; diff --git a/src/lib/jobs/wearable-cycle-repair.ts b/src/lib/jobs/wearable-cycle-repair.ts new file mode 100644 index 000000000..f4ed25108 --- /dev/null +++ b/src/lib/jobs/wearable-cycle-repair.ts @@ -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 { + 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 { + 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 { + const rows = await prisma.$queryRaw(candidateQuery(null)); + const byUser = new Map(); + 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>[], + prisma: PrismaClient = defaultPrisma, +): Promise { + 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(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), + }; + } +} diff --git a/src/lib/oura/__tests__/cycle-sync.test.ts b/src/lib/oura/__tests__/cycle-sync.test.ts index 4319cbfed..ac926f39e 100644 --- a/src/lib/oura/__tests__/cycle-sync.test.ts +++ b/src/lib/oura/__tests__/cycle-sync.test.ts @@ -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" }, diff --git a/src/lib/oura/cycle-sync.ts b/src/lib/oura/cycle-sync.ts index 5a8db8876..b2b33052a 100644 --- a/src/lib/oura/cycle-sync.ts +++ b/src/lib/oura/cycle-sync.ts @@ -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); } diff --git a/tests/integration/__golden__/queue-registration.json b/tests/integration/__golden__/queue-registration.json index 0560c4642..5bb1305d3 100644 --- a/tests/integration/__golden__/queue-registration.json +++ b/tests/integration/__golden__/queue-registration.json @@ -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": { diff --git a/tests/integration/garmin-sync.test.ts b/tests/integration/garmin-sync.test.ts index 0fadd773c..33a572b80 100644 --- a/tests/integration/garmin-sync.test.ts +++ b/tests/integration/garmin-sync.test.ts @@ -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}`, { diff --git a/tests/integration/wearable-cycle-repair.test.ts b/tests/integration/wearable-cycle-repair.test.ts new file mode 100644 index 000000000..b9d95b54a --- /dev/null +++ b/tests/integration/wearable-cycle-repair.test.ts @@ -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); + }); +});