From 7161cbe6c552ff5fed4e9b59521a92d70cb96011 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 26 May 2026 16:07:29 -0700 Subject: [PATCH 1/8] Fix weekly health summary --- .../repositories/weekly-report-repository.ts | 17 +++-- .../server/src/routers/weekly-report.test.ts | 68 +++++++++++++++---- packages/server/src/routers/weekly-report.ts | 40 ++++------- .../src/components/WeeklyReportCard.test.tsx | 25 +++++++ .../web/src/components/WeeklyReportCard.tsx | 27 +++++--- 5 files changed, 120 insertions(+), 57 deletions(-) diff --git a/packages/server/src/repositories/weekly-report-repository.ts b/packages/server/src/repositories/weekly-report-repository.ts index e3a2de6ff1..fa50242aba 100644 --- a/packages/server/src/repositories/weekly-report-repository.ts +++ b/packages/server/src/repositories/weekly-report-repository.ts @@ -12,7 +12,7 @@ import { restingHeartRateClickHouseCte } from "./resting-heart-rate-query.ts"; export type StrainZone = "restoring" | "optimal" | "overreaching"; export interface WeekSummary { - /** ISO week start date (Monday) */ + /** Week start date (Sunday) */ weekStart: string; /** Total training hours */ trainingHours: number; @@ -167,10 +167,13 @@ export class WeeklyReportRepository { WHERE user_id = {userId:UUID} AND toDate(toTimeZone(started_at, {timezone:String})) >= toDate({windowStart:String}) AND ended_at IS NOT NULL - AND avg_hr IS NOT NULL ), daily_training AS ( - SELECT date, sum(hours) AS hours, toInt32(count()) AS count, sum(load) AS load + SELECT + date, + sum(hours) AS hours, + toInt32(count()) AS count, + sumIf(load, load IS NOT NULL) AS load FROM per_activity GROUP BY date ), @@ -216,17 +219,17 @@ export class WeeklyReportRepository { ), weekly AS ( SELECT - toMonday(d.date) AS week_start, + toStartOfWeek(d.date) AS week_start, sum(d.hours) AS total_hours, toInt32(sum(d.count)) AS activity_count, avg(d.load) AS avg_daily_load, avg(sl.duration_minutes) AS avg_sleep_min, - avg(m.resting_hr) AS avg_resting_hr, - avg(m.hrv) AS avg_hrv + avg(nullIf(m.resting_hr, 0)) AS avg_resting_hr, + avg(nullIf(m.hrv, 0)) AS avg_hrv FROM daily d LEFT JOIN sleep_daily sl ON sl.date = d.date LEFT JOIN metrics_daily m ON m.date = d.date - GROUP BY toMonday(d.date) + GROUP BY toStartOfWeek(d.date) ORDER BY week_start ASC ) SELECT diff --git a/packages/server/src/routers/weekly-report.test.ts b/packages/server/src/routers/weekly-report.test.ts index 3cf5edfb51..01a11ed0fe 100644 --- a/packages/server/src/routers/weekly-report.test.ts +++ b/packages/server/src/routers/weekly-report.test.ts @@ -496,10 +496,10 @@ describe("weeklyReportRouter", () => { expect(result.current?.weekStart).toBe("2026-03-17"); }); - it("uses the previous full week as current when the latest requested week is partial", async () => { + it("uses the current Sunday-start week on Tuesday instead of falling back to last week", async () => { const rows = [ { - week_start: "2026-03-09", + week_start: "2026-05-17", total_hours: 4, activity_count: 2, avg_daily_load: 1, @@ -510,9 +510,9 @@ describe("weeklyReportRouter", () => { prev_3wk_avg_sleep: 420, }, { - week_start: "2026-03-16", - total_hours: 1, - activity_count: 1, + week_start: "2026-05-24", + total_hours: 2, + activity_count: 3, avg_daily_load: 0.5, avg_sleep_min: 390, avg_resting_hr: 62, @@ -528,16 +528,18 @@ describe("weeklyReportRouter", () => { sensorStore: makeMockSensorStore(rows), }); - const result = await caller.report({ weeks: 2, endDate: "2026-03-17" }); + const result = await caller.report({ weeks: 2, endDate: "2026-05-26" }); - expect(result.current?.weekStart).toBe("2026-03-09"); - expect(result.history).toEqual([]); + expect(result.current?.weekStart).toBe("2026-05-24"); + expect(result.current?.activityCount).toBe(3); + expect(result.history).toHaveLength(1); + expect(result.history[0]?.weekStart).toBe("2026-05-17"); }); - it("treats Sunday as part of the week that started six days earlier", async () => { + it("groups weekly rows by Sunday-start weeks", async () => { const rows = [ { - week_start: "2026-03-09", + week_start: "2026-03-15", total_hours: 4, activity_count: 2, avg_daily_load: 1, @@ -548,7 +550,7 @@ describe("weeklyReportRouter", () => { prev_3wk_avg_sleep: 420, }, { - week_start: "2026-03-16", + week_start: "2026-03-22", total_hours: 1, activity_count: 1, avg_daily_load: 0.5, @@ -559,17 +561,55 @@ describe("weeklyReportRouter", () => { prev_3wk_avg_sleep: 420, }, ]; + const sensorStore = makeMockSensorStore(rows); const caller = createCaller({ db: { execute: vi.fn().mockResolvedValue(rows) }, userId: "user-1", timezone: "UTC", - sensorStore: makeMockSensorStore(rows), + sensorStore, }); const result = await caller.report({ weeks: 2, endDate: "2026-03-22" }); - expect(result.current?.weekStart).toBe("2026-03-09"); - expect(result.history).toEqual([]); + expect(result.current?.weekStart).toBe("2026-03-22"); + expect(result.history[0]?.weekStart).toBe("2026-03-15"); + const queryText = vi.mocked(sensorStore.query).mock.calls[0]?.[1] ?? ""; + expect(queryText).toContain("toStartOfWeek(d.date)"); + expect(queryText).not.toContain("toMonday(d.date)"); + }); + + it("counts all ended activities while training load only uses rows with average heart rate", async () => { + const sensorStore = makeMockSensorStore([]); + const caller = createCaller({ + db: { execute: vi.fn().mockResolvedValue([]) }, + userId: "user-1", + timezone: "UTC", + sensorStore, + }); + + await caller.report({ weeks: 1, endDate: "2026-05-26" }); + + const queryText = vi.mocked(sensorStore.query).mock.calls[0]?.[1] ?? ""; + expect(queryText).not.toContain("AND avg_hr IS NOT NULL"); + expect(queryText).toContain("* avg_hr / nullIf(toFloat64(max_hr), 0)"); + expect(queryText).toContain("toInt32(count()) AS count"); + expect(queryText).toContain("sumIf(load, load IS NOT NULL) AS load"); + }); + + it("ignores ClickHouse join-default zeros when averaging weekly vitals", async () => { + const sensorStore = makeMockSensorStore([]); + const caller = createCaller({ + db: { execute: vi.fn().mockResolvedValue([]) }, + userId: "user-1", + timezone: "UTC", + sensorStore, + }); + + await caller.report({ weeks: 1, endDate: "2026-05-26" }); + + const queryText = vi.mocked(sensorStore.query).mock.calls[0]?.[1] ?? ""; + expect(queryText).toContain("avg(nullIf(m.resting_hr, 0)) AS avg_resting_hr"); + expect(queryText).toContain("avg(nullIf(m.hrv, 0)) AS avg_hrv"); }); it("verifies full computed values for a single week (kills || 0, rounding, null-check mutants)", async () => { diff --git a/packages/server/src/routers/weekly-report.ts b/packages/server/src/routers/weekly-report.ts index dce04675f2..999040bca2 100644 --- a/packages/server/src/routers/weekly-report.ts +++ b/packages/server/src/routers/weekly-report.ts @@ -23,7 +23,7 @@ function requireSensorStore( export type StrainZone = "restoring" | "optimal" | "overreaching"; export interface WeekSummary { - /** ISO week start date (Monday) */ + /** Week start date (Sunday) */ weekStart: string; /** Total training hours */ trainingHours: number; @@ -67,7 +67,7 @@ export function classifyStrainZone(weekAvgLoad: number, chronicAvgLoad: number): export const weeklyReportRouter = router({ /** * Weekly Performance Report — mirrors Whoop's Weekly Performance Assessment. - * Aggregates strain balance, sleep performance, readiness, and key vitals per ISO week. + * Aggregates strain balance, sleep performance, readiness, and key vitals per Sunday-start week. */ report: cachedProtectedQuery(CacheTTL.LONG) .input(z.object({ weeks: z.number().min(1).max(52).default(12), endDate: endDateSchema })) @@ -101,10 +101,13 @@ export const weeklyReportRouter = router({ WHERE user_id = {userId:UUID} AND toDate(toTimeZone(started_at, {timezone:String})) >= toDate({windowStart:String}) AND ended_at IS NOT NULL - AND avg_hr IS NOT NULL ), daily_training AS ( - SELECT date, sum(hours) AS hours, toInt32(count()) AS count, sum(load) AS load + SELECT + date, + sum(hours) AS hours, + toInt32(count()) AS count, + sumIf(load, load IS NOT NULL) AS load FROM per_activity GROUP BY date ), @@ -148,17 +151,17 @@ export const weeklyReportRouter = router({ ), weekly AS ( SELECT - toMonday(d.date) AS week_start, + toStartOfWeek(d.date) AS week_start, sum(d.hours) AS total_hours, toInt32(sum(d.count)) AS activity_count, avg(d.load) AS avg_daily_load, avg(sl.duration_minutes) AS avg_sleep_min, - avg(m.resting_hr) AS avg_resting_hr, - avg(m.hrv) AS avg_hrv + avg(nullIf(m.resting_hr, 0)) AS avg_resting_hr, + avg(nullIf(m.hrv, 0)) AS avg_hrv FROM daily d LEFT JOIN sleep_daily sl ON sl.date = d.date LEFT JOIN metrics_daily m ON m.date = d.date - GROUP BY toMonday(d.date) + GROUP BY toStartOfWeek(d.date) ORDER BY week_start ASC ) SELECT @@ -204,27 +207,12 @@ export const weeklyReportRouter = router({ } satisfies WeekSummary; }); - // Only return the requested number of weeks. - // If the most recent week just started (today is its first day), it has - // no meaningful data yet — treat the previous full week as "current". + // Only return the requested number of Sunday-start weeks. const cutoffWeeks = parsed.slice(-input.weeks); const lastWeek = cutoffWeeks[cutoffWeeks.length - 1]; - const isPartialWeek = lastWeek && lastWeek.weekStart === getMondayOfWeek(input.endDate); - const current = - isPartialWeek && cutoffWeeks.length >= 2 - ? (cutoffWeeks[cutoffWeeks.length - 2] ?? null) - : (lastWeek ?? null); - const history = isPartialWeek ? cutoffWeeks.slice(0, -2) : cutoffWeeks.slice(0, -1); + const current = lastWeek ?? null; + const history = cutoffWeeks.slice(0, -1); return { current, history }; }), }); - -/** Return the ISO Monday (YYYY-MM-DD) for the week containing `dateStr`. */ -function getMondayOfWeek(dateStr: string): string { - const date = new Date(`${dateStr}T12:00:00Z`); - const day = date.getUTCDay(); // 0=Sun, 1=Mon, ... - const diff = day === 0 ? 6 : day - 1; // days since Monday - date.setUTCDate(date.getUTCDate() - diff); - return date.toISOString().slice(0, 10); -} diff --git a/packages/web/src/components/WeeklyReportCard.test.tsx b/packages/web/src/components/WeeklyReportCard.test.tsx index e8e1a7bf7d..15bca3e178 100644 --- a/packages/web/src/components/WeeklyReportCard.test.tsx +++ b/packages/web/src/components/WeeklyReportCard.test.tsx @@ -29,4 +29,29 @@ describe("WeeklyReportCard", () => { expect(screen.queryByText("Overreaching")).toBeNull(); expect(screen.getByText("Not tracked")).toBeTruthy(); }); + + it("shows no-training status instead of optimal when the week has no activities", () => { + render( + , + ); + + expect(screen.getByText("No training")).toBeTruthy(); + expect(screen.queryByText("Optimal")).toBeNull(); + }); }); diff --git a/packages/web/src/components/WeeklyReportCard.tsx b/packages/web/src/components/WeeklyReportCard.tsx index ab9208a16d..3b1084830c 100644 --- a/packages/web/src/components/WeeklyReportCard.tsx +++ b/packages/web/src/components/WeeklyReportCard.tsx @@ -25,6 +25,7 @@ export function WeeklyReportCard({ data, loading }: WeeklyReportCardProps) { const currentZone = new StrainZone(current.strainZone); const zoneColor = currentZone.color; const sleepWasTracked = current.avgSleepMinutes > 0; + const hasTraining = current.activityCount > 0 || current.trainingHours > 0; const prevWeek = history.length > 0 ? history[history.length - 1] : null; return ( @@ -34,17 +35,21 @@ export function WeeklyReportCard({ data, loading }: WeeklyReportCardProps) {

Weekly Performance

Week of {formatDateShort(current.weekStart)}

- {sleepWasTracked ? ( + {!sleepWasTracked ? ( +
+ Sleep not tracked +
+ ) : !hasTraining ? ( +
+ No training +
+ ) : (
{currentZone.label}
- ) : ( -
- Sleep not tracked -
)} @@ -93,16 +98,18 @@ export function WeeklyReportCard({ data, loading }: WeeklyReportCardProps) { {history.slice(-8).map((w) => { const zone = new StrainZone(w.strainZone); const weekHasSleepData = w.avgSleepMinutes > 0; + const weekHasTraining = w.activityCount > 0 || w.trainingHours > 0; + const showZone = weekHasSleepData && weekHasTraining; return (
); })} - {sleepWasTracked ? ( + {sleepWasTracked && hasTraining ? (
)}
From 221c26662961cd2daf5c133624fbab05dcb2e5b5 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 26 May 2026 16:24:03 -0700 Subject: [PATCH 2/8] Address weekly report review --- packages/server/src/repositories/weekly-report-repository.ts | 4 ++-- packages/server/src/routers/weekly-report.test.ts | 2 +- packages/server/src/routers/weekly-report.ts | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/server/src/repositories/weekly-report-repository.ts b/packages/server/src/repositories/weekly-report-repository.ts index fa50242aba..ddac3cf3bf 100644 --- a/packages/server/src/repositories/weekly-report-repository.ts +++ b/packages/server/src/repositories/weekly-report-repository.ts @@ -219,7 +219,7 @@ export class WeeklyReportRepository { ), weekly AS ( SELECT - toStartOfWeek(d.date) AS week_start, + toStartOfWeek(d.date, 1) AS week_start, sum(d.hours) AS total_hours, toInt32(sum(d.count)) AS activity_count, avg(d.load) AS avg_daily_load, @@ -229,7 +229,7 @@ export class WeeklyReportRepository { FROM daily d LEFT JOIN sleep_daily sl ON sl.date = d.date LEFT JOIN metrics_daily m ON m.date = d.date - GROUP BY toStartOfWeek(d.date) + GROUP BY toStartOfWeek(d.date, 1) ORDER BY week_start ASC ) SELECT diff --git a/packages/server/src/routers/weekly-report.test.ts b/packages/server/src/routers/weekly-report.test.ts index 01a11ed0fe..85e9913a66 100644 --- a/packages/server/src/routers/weekly-report.test.ts +++ b/packages/server/src/routers/weekly-report.test.ts @@ -574,7 +574,7 @@ describe("weeklyReportRouter", () => { expect(result.current?.weekStart).toBe("2026-03-22"); expect(result.history[0]?.weekStart).toBe("2026-03-15"); const queryText = vi.mocked(sensorStore.query).mock.calls[0]?.[1] ?? ""; - expect(queryText).toContain("toStartOfWeek(d.date)"); + expect(queryText).toContain("toStartOfWeek(d.date, 1)"); expect(queryText).not.toContain("toMonday(d.date)"); }); diff --git a/packages/server/src/routers/weekly-report.ts b/packages/server/src/routers/weekly-report.ts index 999040bca2..4a34ae099f 100644 --- a/packages/server/src/routers/weekly-report.ts +++ b/packages/server/src/routers/weekly-report.ts @@ -151,7 +151,7 @@ export const weeklyReportRouter = router({ ), weekly AS ( SELECT - toStartOfWeek(d.date) AS week_start, + toStartOfWeek(d.date, 1) AS week_start, sum(d.hours) AS total_hours, toInt32(sum(d.count)) AS activity_count, avg(d.load) AS avg_daily_load, @@ -161,7 +161,7 @@ export const weeklyReportRouter = router({ FROM daily d LEFT JOIN sleep_daily sl ON sl.date = d.date LEFT JOIN metrics_daily m ON m.date = d.date - GROUP BY toStartOfWeek(d.date) + GROUP BY toStartOfWeek(d.date, 1) ORDER BY week_start ASC ) SELECT From be3dc1bb0f69b84274618fc05c3b9c54bc6fb3c3 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 26 May 2026 16:34:45 -0700 Subject: [PATCH 3/8] Fix weekly report Sunday bucketing --- packages/server/src/repositories/weekly-report-repository.ts | 4 ++-- packages/server/src/routers/weekly-report.test.ts | 2 +- packages/server/src/routers/weekly-report.ts | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/server/src/repositories/weekly-report-repository.ts b/packages/server/src/repositories/weekly-report-repository.ts index ddac3cf3bf..1cc1dc7404 100644 --- a/packages/server/src/repositories/weekly-report-repository.ts +++ b/packages/server/src/repositories/weekly-report-repository.ts @@ -219,7 +219,7 @@ export class WeeklyReportRepository { ), weekly AS ( SELECT - toStartOfWeek(d.date, 1) AS week_start, + toStartOfWeek(d.date, 0) AS week_start, sum(d.hours) AS total_hours, toInt32(sum(d.count)) AS activity_count, avg(d.load) AS avg_daily_load, @@ -229,7 +229,7 @@ export class WeeklyReportRepository { FROM daily d LEFT JOIN sleep_daily sl ON sl.date = d.date LEFT JOIN metrics_daily m ON m.date = d.date - GROUP BY toStartOfWeek(d.date, 1) + GROUP BY toStartOfWeek(d.date, 0) ORDER BY week_start ASC ) SELECT diff --git a/packages/server/src/routers/weekly-report.test.ts b/packages/server/src/routers/weekly-report.test.ts index 85e9913a66..743c80e5f4 100644 --- a/packages/server/src/routers/weekly-report.test.ts +++ b/packages/server/src/routers/weekly-report.test.ts @@ -574,7 +574,7 @@ describe("weeklyReportRouter", () => { expect(result.current?.weekStart).toBe("2026-03-22"); expect(result.history[0]?.weekStart).toBe("2026-03-15"); const queryText = vi.mocked(sensorStore.query).mock.calls[0]?.[1] ?? ""; - expect(queryText).toContain("toStartOfWeek(d.date, 1)"); + expect(queryText).toContain("toStartOfWeek(d.date, 0)"); expect(queryText).not.toContain("toMonday(d.date)"); }); diff --git a/packages/server/src/routers/weekly-report.ts b/packages/server/src/routers/weekly-report.ts index 4a34ae099f..0d00ba8b23 100644 --- a/packages/server/src/routers/weekly-report.ts +++ b/packages/server/src/routers/weekly-report.ts @@ -151,7 +151,7 @@ export const weeklyReportRouter = router({ ), weekly AS ( SELECT - toStartOfWeek(d.date, 1) AS week_start, + toStartOfWeek(d.date, 0) AS week_start, sum(d.hours) AS total_hours, toInt32(sum(d.count)) AS activity_count, avg(d.load) AS avg_daily_load, @@ -161,7 +161,7 @@ export const weeklyReportRouter = router({ FROM daily d LEFT JOIN sleep_daily sl ON sl.date = d.date LEFT JOIN metrics_daily m ON m.date = d.date - GROUP BY toStartOfWeek(d.date, 1) + GROUP BY toStartOfWeek(d.date, 0) ORDER BY week_start ASC ) SELECT From 56257f5203e4b3797baec1c83d46052c9096e91b Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 26 May 2026 17:54:17 -0700 Subject: [PATCH 4/8] Fix analytics worker activity microbatches --- analytics/macros/bounded_activity_graph.sql | 206 ++++++++++++++++++ .../read_models/activity_location_sample.sql | 8 +- .../read_models/activity_sensor_sample.sql | 11 +- .../read_model_microbatch.sql.test.ts | 27 ++- analytics/models/sources.yml | 2 + docs/production-incident-baseline.md | 41 ++++ entrypoint.sh | 1 - 7 files changed, 277 insertions(+), 19 deletions(-) create mode 100644 analytics/macros/bounded_activity_graph.sql diff --git a/analytics/macros/bounded_activity_graph.sql b/analytics/macros/bounded_activity_graph.sql new file mode 100644 index 0000000000..661c718c05 --- /dev/null +++ b/analytics/macros/bounded_activity_graph.sql @@ -0,0 +1,206 @@ +{% macro bounded_activity_graph() %} +{% set batch_start_value = config.get("__dbt_internal_microbatch_event_time_start") %} +{% set batch_end_value = config.get("__dbt_internal_microbatch_event_time_end") %} +{% if batch_start_value and batch_end_value %} +{% set batch_start = batch_start_value.strftime("%Y-%m-%d %H:%M:%S") %} +{% set batch_end = batch_end_value.strftime("%Y-%m-%d %H:%M:%S") %} +{% endif %} +active_activity AS ( + SELECT * + FROM {{ source('postgres_fitness', 'activity') }} FINAL + WHERE _peerdb_is_deleted = 0 + {% if batch_start_value and batch_end_value %} + AND started_at < toDateTime64('{{ batch_end }}', 6, 'UTC') + AND coalesce(ended_at, started_at + INTERVAL 12 HOUR) >= toDateTime64('{{ batch_start }}', 6, 'UTC') + {% endif %} +), + +active_provider_priority AS ( + SELECT * + FROM {{ source('postgres_fitness', 'provider_priority') }} FINAL + WHERE _peerdb_is_deleted = 0 +), + +active_device_priority AS ( + SELECT * + FROM {{ source('postgres_fitness', 'device_priority') }} FINAL + WHERE _peerdb_is_deleted = 0 +), + +device_priority_match AS ( + SELECT + activity_id, + priority + FROM ( + SELECT + active_activity.id AS activity_id, + active_device_priority.priority AS priority, + row_number() OVER ( + PARTITION BY active_activity.id + ORDER BY length(active_device_priority.source_name_pattern) DESC + ) AS row_number + FROM active_activity + INNER JOIN active_device_priority + ON active_device_priority.provider_id = active_activity.provider_id + AND active_activity.source_name LIKE active_device_priority.source_name_pattern + ) + WHERE row_number = 1 +), + +ranked AS ( + SELECT + active_activity.id AS id, + active_activity.provider_id AS provider_id, + active_activity.user_id AS user_id, + active_activity.external_id AS external_id, + active_activity.activity_type AS activity_type, + active_activity.started_at AS started_at, + active_activity.ended_at AS ended_at, + active_activity.source_name AS source_name, + active_activity.name AS name, + active_activity.notes AS notes, + active_activity.timezone AS timezone, + active_activity.raw AS raw, + coalesce(device_priority_match.priority, active_provider_priority.priority, 100) AS priority + FROM active_activity + LEFT JOIN active_provider_priority + ON active_provider_priority.provider_id = active_activity.provider_id + LEFT JOIN device_priority_match + ON device_priority_match.activity_id = active_activity.id +), + +pairs AS ( + SELECT + left_activity.id AS id1, + right_activity.id AS id2 + FROM ranked AS left_activity + INNER JOIN ranked AS right_activity + ON left_activity.user_id = right_activity.user_id + AND toString(left_activity.id) < toString(right_activity.id) + AND dateDiff( + 'second', + greatest(left_activity.started_at, right_activity.started_at), + least( + coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 1 HOUR), + coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 1 HOUR) + ) + ) / nullIf(dateDiff( + 'second', + least(left_activity.started_at, right_activity.started_at), + greatest( + coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 1 HOUR), + coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 1 HOUR) + ) + ), 0) > 0.8 +), + +graph_edges AS ( + SELECT + id1 AS from_id, + id2 AS to_id + FROM pairs + UNION ALL + SELECT + id2 AS from_id, + id1 AS to_id + FROM pairs +), + +connected_components AS ( + SELECT + id AS activity_id, + id AS connected_activity_id, + [toString(id)] AS visited_activity_ids + FROM ranked + UNION ALL + SELECT + connected_components.activity_id AS activity_id, + graph_edges.to_id AS connected_activity_id, + arrayConcat(connected_components.visited_activity_ids, [toString(graph_edges.to_id)]) AS visited_activity_ids + FROM connected_components + INNER JOIN graph_edges + ON graph_edges.from_id = connected_components.connected_activity_id + WHERE NOT has(connected_components.visited_activity_ids, toString(graph_edges.to_id)) +), + +final_groups AS ( + SELECT + activity_id, + min(toString(connected_activity_id)) AS group_id + FROM connected_components + GROUP BY activity_id +), + +best AS ( + SELECT * + FROM ( + SELECT + final_groups.group_id AS group_id, + ranked.id AS canonical_id, + ranked.provider_id AS provider_id, + ranked.user_id AS user_id, + ranked.activity_type AS activity_type, + ranked.started_at AS started_at, + ranked.ended_at AS ended_at, + ranked.source_name AS source_name, + ranked.priority AS priority, + row_number() OVER ( + PARTITION BY final_groups.group_id + ORDER BY ranked.priority ASC, toString(ranked.id) ASC + ) AS row_number + FROM final_groups + INNER JOIN ranked + ON ranked.id = final_groups.activity_id + ) + WHERE row_number = 1 +), + +merged AS ( + SELECT + best.group_id AS group_id, + best.canonical_id AS id, + any(best.provider_id) AS provider_id, + any(best.user_id) AS user_id, + any(best.activity_type) AS activity_type, + any(best.started_at) AS started_at, + any(best.ended_at) AS ended_at, + any(best.source_name) AS source_name, + argMinIf(ranked.name, ranked.priority, ranked.name IS NOT NULL) AS name, + argMinIf(ranked.notes, ranked.priority, ranked.notes IS NOT NULL) AS notes, + argMinIf(ranked.timezone, ranked.priority, ranked.timezone IS NOT NULL) AS timezone, + argMinIf(ranked.raw, ranked.priority, ranked.raw IS NOT NULL) AS raw, + arraySort(groupUniqArray(ranked.provider_id)) AS source_providers, + groupArrayIf( + map('providerId', ranked.provider_id, 'externalId', ranked.external_id), + ranked.external_id IS NOT NULL AND ranked.external_id != '' + ) AS source_external_ids, + groupArray(ranked.id) AS member_activity_ids + FROM best + INNER JOIN final_groups + ON final_groups.group_id = best.group_id + INNER JOIN ranked + ON ranked.id = final_groups.activity_id + GROUP BY best.group_id, best.canonical_id +), + +current_activity AS ( + SELECT + id AS activity_id, + user_id, + activity_type, + name, + started_at, + ended_at + FROM merged +), + +activity_members AS ( + SELECT + id AS activity_id, + user_id, + started_at, + ended_at, + arrayJoin(member_activity_ids) AS member_activity_id + FROM merged +) +{% endmacro %} diff --git a/analytics/models/read_models/activity_location_sample.sql b/analytics/models/read_models/activity_location_sample.sql index bec1c41f27..90c7e15cef 100644 --- a/analytics/models/read_models/activity_location_sample.sql +++ b/analytics/models/read_models/activity_location_sample.sql @@ -15,13 +15,7 @@ } ) }} -WITH activity_members AS ( - SELECT - activity_id, - user_id, - member_activity_id - FROM {{ source('analytics', 'v_activity_members') }} -), +WITH RECURSIVE {{ bounded_activity_graph() }}, location_versions AS ( SELECT * diff --git a/analytics/models/read_models/activity_sensor_sample.sql b/analytics/models/read_models/activity_sensor_sample.sql index f3959062a0..f8480d7cec 100644 --- a/analytics/models/read_models/activity_sensor_sample.sql +++ b/analytics/models/read_models/activity_sensor_sample.sql @@ -15,14 +15,7 @@ } ) }} -WITH current_activity AS ( - SELECT - id AS activity_id, - user_id, - started_at, - coalesce(ended_at, started_at + INTERVAL 12 HOUR) AS ended_at - FROM {{ source('analytics', 'v_activity') }} -), +WITH RECURSIVE {{ bounded_activity_graph() }}, activity_samples AS ( SELECT @@ -37,7 +30,7 @@ activity_samples AS ( INNER JOIN current_activity ON current_activity.user_id = samples.user_id AND samples.recorded_at >= current_activity.started_at - AND samples.recorded_at <= current_activity.ended_at + AND samples.recorded_at <= coalesce(current_activity.ended_at, current_activity.started_at + INTERVAL 12 HOUR) ) SELECT diff --git a/analytics/models/read_models/read_model_microbatch.sql.test.ts b/analytics/models/read_models/read_model_microbatch.sql.test.ts index 96bcd551ed..a70539f700 100644 --- a/analytics/models/read_models/read_model_microbatch.sql.test.ts +++ b/analytics/models/read_models/read_model_microbatch.sql.test.ts @@ -14,6 +14,14 @@ function compactWhitespace(value: string): string { } describe("production analytics read-model build", () => { + it("does not block the BullMQ worker on analytics dbt builds", () => { + const entrypoint = readProjectFile("entrypoint.sh"); + const workerBlockMatch = entrypoint.match(/ worker\)\n(?[\s\S]*?)\n ;;/); + + expect(workerBlockMatch?.groups?.body).toContain("exec $NODE src/jobs/worker.ts"); + expect(workerBlockMatch?.groups?.body).not.toContain("dbt build"); + }); + it("runs every bounded intermediary and final read model in dependency order", () => { const entrypoint = readProjectFile("entrypoint.sh"); const safeModelMatch = entrypoint.match(/^DBT_SAFE_MODELS="([^"]+)"$/m); @@ -60,7 +68,9 @@ describe("production analytics read-model build", () => { expect(sql).toContain("incremental_strategy='microbatch'"); expect(sql).toContain("event_time='recorded_at'"); expect(sql).toContain("ref('deduped_sensor')"); - expect(sql).toContain("source('analytics', 'v_activity')"); + expect(sql).toContain("WITH RECURSIVE {{ bounded_activity_graph() }}"); + expect(sql).toContain("bounded_activity_graph()"); + expect(sql).not.toContain("source('analytics', 'v_activity')"); expect(sql).toContain("activity_id"); }); @@ -71,10 +81,23 @@ describe("production analytics read-model build", () => { expect(sql).toContain("incremental_strategy='microbatch'"); expect(sql).toContain("event_time='recorded_at'"); expect(sql).toContain("source('postgres_fitness', 'metric_stream')"); - expect(sql).toContain("source('analytics', 'v_activity_members')"); + expect(sql).toContain("WITH RECURSIVE {{ bounded_activity_graph() }}"); + expect(sql).toContain("bounded_activity_graph()"); + expect(sql).not.toContain("source('analytics', 'v_activity_members')"); expect(sql).toContain("channel = 'location'"); }); + it("bounds activity graph construction to the active microbatch window", () => { + const sql = readProjectFile("analytics/macros/bounded_activity_graph.sql"); + + expect(sql).toContain("__dbt_internal_microbatch_event_time_start"); + expect(sql).toContain("__dbt_internal_microbatch_event_time_end"); + expect(sql).toContain("started_at < toDateTime64('{{ batch_end }}', 6, 'UTC')"); + expect(sql).toContain( + "coalesce(ended_at, started_at + INTERVAL 12 HOUR) >= toDateTime64('{{ batch_start }}', 6, 'UTC')", + ); + }); + it("aggregates activity sensor summary from the bounded sensor intermediary", () => { expect(existsSync(new URL("./activity_sensor_summary_rows.sql", import.meta.url))).toBe(true); const sql = readModel("activity_sensor_summary_rows"); diff --git a/analytics/models/sources.yml b/analytics/models/sources.yml index 5ee4f8e7d1..817b42c831 100644 --- a/analytics/models/sources.yml +++ b/analytics/models/sources.yml @@ -10,9 +10,11 @@ sources: schema: postgres_fitness tables: - name: activity + - name: device_priority - name: metric_stream config: event_time: recorded_at + - name: provider_priority - name: sensor_device_priority - name: sensor_provider_priority - name: sleep_session diff --git a/docs/production-incident-baseline.md b/docs/production-incident-baseline.md index 4d077e1917..d844b16069 100644 --- a/docs/production-incident-baseline.md +++ b/docs/production-incident-baseline.md @@ -8796,3 +8796,44 @@ new incremental tables are populated. - Follow-Up Work: Deploy the prepared bounded RHR and activity summary model update; add freshness monitoring comparing latest sleep/heart-rate inputs to latest active RHR output date. + +### Production worker restart loop from analytics dbt OOM + +- Date: 2026-05-26 PT / 2026-05-27 UTC. +- Symptoms: Production was reported down. Public `/healthz` and the SPA shell + still returned HTTP 200, but background processing was impaired: the + `dofek_worker` service was repeatedly failing with `task: non-zero exit (1)`, + and `dofek_analytics-worker` was retrying dbt builds. +- User Impact: Background sync and import jobs were unavailable while the + worker was in a restart loop. ClickHouse-backed dashboard data was at risk of + stale or failed reads while ClickHouse was under memory pressure. +- Evidence: The first fatal analytics-worker log lines were ClickHouse + `MEMORY_LIMIT_EXCEEDED` exceptions in dbt models including + `activity_location_sample` and `activity_sensor_sample`, with ClickHouse at + its configured 3 GiB memory limit while executing recursive activity graph + work. The failing batch for `2026-05-20` had only 458 heart-rate sample rows + but used `analytics.v_activity`/`analytics.v_activity_members`, which forced + ClickHouse to traverse the unbounded 1,029-activity user graph before the + microbatch could finish. The worker logs also showed the `worker` entrypoint + running Postgres migrations, ClickHouse migrations, and then `dbt build` + before the BullMQ worker process could start. +- Root Cause: The activity sample dbt models used global recursive activity + views inside each microbatch. ClickHouse did not push the day filter inside + those views, so each batch recomputed the full dedupe graph and OOMed. The + blast radius was larger because `entrypoint.sh` also coupled the BullMQ + `worker` service startup to the same analytics dbt build. +- Fix/Mitigation: A short-lived manual mitigation stopped the failing dbt retry + loop and temporarily ran the Node BullMQ worker directly so background jobs + could resume. The durable code change removes `dbt build` from the `worker` + entrypoint and bounds the activity graph inside the affected analytics models + to the active microbatch window so `analytics-worker` remains enabled. +- Validation: After the current production image was redeployed, + `dofek_web` was `2/2`, `dofek_worker` was `1/1`, and + `dofek_analytics-worker` was `1/1`. `/healthz` returned HTTP 200. The local + fix compiles the affected dbt models with batch-local activity graph bounds + and passes the changed test suite. +- Remaining Risk: Medium until the bounded activity graph fix is deployed and + the next scheduled analytics-worker run completes cleanly on production-scale + data. +- Follow-Up Work: Add freshness monitoring for analytics read models and keep + the BullMQ worker independent from analytics rebuilds. diff --git a/entrypoint.sh b/entrypoint.sh index 53d60dc4d9..8fdf34e624 100755 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -31,7 +31,6 @@ case "${1:-sync}" in ;; worker) $NODE src/db/run-migrate.ts - dbt build --project-dir analytics --profiles-dir analytics --threads 1 --select $DBT_SAFE_MODELS exec $NODE src/jobs/worker.ts ;; migrate) From 09b94b90ee52aa4fc8cc2462213d6ef24da9fa82 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 26 May 2026 18:40:31 -0700 Subject: [PATCH 5/8] Stabilize Ultrahuman sync integration test --- src/providers/ultrahuman-sync.integration.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/providers/ultrahuman-sync.integration.test.ts b/src/providers/ultrahuman-sync.integration.test.ts index 3cbb373307..6d17dcd00d 100644 --- a/src/providers/ultrahuman-sync.integration.test.ts +++ b/src/providers/ultrahuman-sync.integration.test.ts @@ -1,7 +1,7 @@ import { eq } from "drizzle-orm"; import { HttpResponse, http } from "msw"; import { setupServer } from "msw/node"; -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { dailyMetrics, sleepSession } from "../db/schema.ts"; import { setupTestDatabase, type TestContext } from "../db/test-helpers.ts"; import { ensureProvider, saveTokens } from "../db/tokens.ts"; @@ -124,6 +124,7 @@ describe("UltrahumanProvider.sync() (integration)", () => { "Ultrahuman", "https://partner.ultrahuman.com/api/v1", ); + vi.useFakeTimers({ now: new Date("2026-03-15T12:00:00Z"), toFake: ["Date"] }); }, 60_000); afterEach(() => { @@ -132,6 +133,7 @@ describe("UltrahumanProvider.sync() (integration)", () => { afterAll(async () => { server.close(); + vi.useRealTimers(); if (ctx) await ctx.cleanup(); }); From cd08075c8fc7e15f9b525c6bb4fc4b181147f6da Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 26 May 2026 18:46:03 -0700 Subject: [PATCH 6/8] Use Sunday week mode in weekly reports --- packages/server/src/repositories/weekly-report-repository.ts | 4 ++-- packages/server/src/routers/weekly-report.test.ts | 2 +- packages/server/src/routers/weekly-report.ts | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/server/src/repositories/weekly-report-repository.ts b/packages/server/src/repositories/weekly-report-repository.ts index 1cc1dc7404..ddac3cf3bf 100644 --- a/packages/server/src/repositories/weekly-report-repository.ts +++ b/packages/server/src/repositories/weekly-report-repository.ts @@ -219,7 +219,7 @@ export class WeeklyReportRepository { ), weekly AS ( SELECT - toStartOfWeek(d.date, 0) AS week_start, + toStartOfWeek(d.date, 1) AS week_start, sum(d.hours) AS total_hours, toInt32(sum(d.count)) AS activity_count, avg(d.load) AS avg_daily_load, @@ -229,7 +229,7 @@ export class WeeklyReportRepository { FROM daily d LEFT JOIN sleep_daily sl ON sl.date = d.date LEFT JOIN metrics_daily m ON m.date = d.date - GROUP BY toStartOfWeek(d.date, 0) + GROUP BY toStartOfWeek(d.date, 1) ORDER BY week_start ASC ) SELECT diff --git a/packages/server/src/routers/weekly-report.test.ts b/packages/server/src/routers/weekly-report.test.ts index 743c80e5f4..85e9913a66 100644 --- a/packages/server/src/routers/weekly-report.test.ts +++ b/packages/server/src/routers/weekly-report.test.ts @@ -574,7 +574,7 @@ describe("weeklyReportRouter", () => { expect(result.current?.weekStart).toBe("2026-03-22"); expect(result.history[0]?.weekStart).toBe("2026-03-15"); const queryText = vi.mocked(sensorStore.query).mock.calls[0]?.[1] ?? ""; - expect(queryText).toContain("toStartOfWeek(d.date, 0)"); + expect(queryText).toContain("toStartOfWeek(d.date, 1)"); expect(queryText).not.toContain("toMonday(d.date)"); }); diff --git a/packages/server/src/routers/weekly-report.ts b/packages/server/src/routers/weekly-report.ts index 0d00ba8b23..4a34ae099f 100644 --- a/packages/server/src/routers/weekly-report.ts +++ b/packages/server/src/routers/weekly-report.ts @@ -151,7 +151,7 @@ export const weeklyReportRouter = router({ ), weekly AS ( SELECT - toStartOfWeek(d.date, 0) AS week_start, + toStartOfWeek(d.date, 1) AS week_start, sum(d.hours) AS total_hours, toInt32(sum(d.count)) AS activity_count, avg(d.load) AS avg_daily_load, @@ -161,7 +161,7 @@ export const weeklyReportRouter = router({ FROM daily d LEFT JOIN sleep_daily sl ON sl.date = d.date LEFT JOIN metrics_daily m ON m.date = d.date - GROUP BY toStartOfWeek(d.date, 0) + GROUP BY toStartOfWeek(d.date, 1) ORDER BY week_start ASC ) SELECT From e08e2691637efa5af3b813a30adcf30490630021 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 26 May 2026 18:51:34 -0700 Subject: [PATCH 7/8] Restore Sunday week mode in weekly reports --- packages/server/src/repositories/weekly-report-repository.ts | 4 ++-- packages/server/src/routers/weekly-report.test.ts | 2 +- packages/server/src/routers/weekly-report.ts | 4 ++-- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/server/src/repositories/weekly-report-repository.ts b/packages/server/src/repositories/weekly-report-repository.ts index ddac3cf3bf..1cc1dc7404 100644 --- a/packages/server/src/repositories/weekly-report-repository.ts +++ b/packages/server/src/repositories/weekly-report-repository.ts @@ -219,7 +219,7 @@ export class WeeklyReportRepository { ), weekly AS ( SELECT - toStartOfWeek(d.date, 1) AS week_start, + toStartOfWeek(d.date, 0) AS week_start, sum(d.hours) AS total_hours, toInt32(sum(d.count)) AS activity_count, avg(d.load) AS avg_daily_load, @@ -229,7 +229,7 @@ export class WeeklyReportRepository { FROM daily d LEFT JOIN sleep_daily sl ON sl.date = d.date LEFT JOIN metrics_daily m ON m.date = d.date - GROUP BY toStartOfWeek(d.date, 1) + GROUP BY toStartOfWeek(d.date, 0) ORDER BY week_start ASC ) SELECT diff --git a/packages/server/src/routers/weekly-report.test.ts b/packages/server/src/routers/weekly-report.test.ts index 85e9913a66..743c80e5f4 100644 --- a/packages/server/src/routers/weekly-report.test.ts +++ b/packages/server/src/routers/weekly-report.test.ts @@ -574,7 +574,7 @@ describe("weeklyReportRouter", () => { expect(result.current?.weekStart).toBe("2026-03-22"); expect(result.history[0]?.weekStart).toBe("2026-03-15"); const queryText = vi.mocked(sensorStore.query).mock.calls[0]?.[1] ?? ""; - expect(queryText).toContain("toStartOfWeek(d.date, 1)"); + expect(queryText).toContain("toStartOfWeek(d.date, 0)"); expect(queryText).not.toContain("toMonday(d.date)"); }); diff --git a/packages/server/src/routers/weekly-report.ts b/packages/server/src/routers/weekly-report.ts index 4a34ae099f..0d00ba8b23 100644 --- a/packages/server/src/routers/weekly-report.ts +++ b/packages/server/src/routers/weekly-report.ts @@ -151,7 +151,7 @@ export const weeklyReportRouter = router({ ), weekly AS ( SELECT - toStartOfWeek(d.date, 1) AS week_start, + toStartOfWeek(d.date, 0) AS week_start, sum(d.hours) AS total_hours, toInt32(sum(d.count)) AS activity_count, avg(d.load) AS avg_daily_load, @@ -161,7 +161,7 @@ export const weeklyReportRouter = router({ FROM daily d LEFT JOIN sleep_daily sl ON sl.date = d.date LEFT JOIN metrics_daily m ON m.date = d.date - GROUP BY toStartOfWeek(d.date, 1) + GROUP BY toStartOfWeek(d.date, 0) ORDER BY week_start ASC ) SELECT From 8eb7a8e04f6fbb92a0aa65974fe3eaf6c10bc80b Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 26 May 2026 18:53:10 -0700 Subject: [PATCH 8/8] Align bounded activity graph windows --- analytics/macros/bounded_activity_graph.sql | 12 ++++++------ .../read_model_microbatch.sql.test.ts | 17 +++++++++++++++++ 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/analytics/macros/bounded_activity_graph.sql b/analytics/macros/bounded_activity_graph.sql index 661c718c05..bff2e7f210 100644 --- a/analytics/macros/bounded_activity_graph.sql +++ b/analytics/macros/bounded_activity_graph.sql @@ -81,15 +81,15 @@ pairs AS ( 'second', greatest(left_activity.started_at, right_activity.started_at), least( - coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 1 HOUR), - coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 1 HOUR) + coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 12 HOUR), + coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 12 HOUR) ) ) / nullIf(dateDiff( 'second', least(left_activity.started_at, right_activity.started_at), greatest( - coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 1 HOUR), - coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 1 HOUR) + coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 12 HOUR), + coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 12 HOUR) ) ), 0) > 0.8 ), @@ -162,8 +162,8 @@ merged AS ( any(best.provider_id) AS provider_id, any(best.user_id) AS user_id, any(best.activity_type) AS activity_type, - any(best.started_at) AS started_at, - any(best.ended_at) AS ended_at, + min(ranked.started_at) AS started_at, + max(coalesce(ranked.ended_at, ranked.started_at + INTERVAL 12 HOUR)) AS ended_at, any(best.source_name) AS source_name, argMinIf(ranked.name, ranked.priority, ranked.name IS NOT NULL) AS name, argMinIf(ranked.notes, ranked.priority, ranked.notes IS NOT NULL) AS notes, diff --git a/analytics/models/read_models/read_model_microbatch.sql.test.ts b/analytics/models/read_models/read_model_microbatch.sql.test.ts index a70539f700..0c2820690c 100644 --- a/analytics/models/read_models/read_model_microbatch.sql.test.ts +++ b/analytics/models/read_models/read_model_microbatch.sql.test.ts @@ -98,6 +98,23 @@ describe("production analytics read-model build", () => { ); }); + it("uses the same null-ended activity window for overlap matching", () => { + const sql = readProjectFile("analytics/macros/bounded_activity_graph.sql"); + + expect(sql).toContain("coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 12 HOUR)"); + expect(sql).toContain("coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 12 HOUR)"); + expect(sql).not.toContain("INTERVAL 1 HOUR"); + }); + + it("uses merged group time bounds for current activity sensor membership", () => { + const sql = readProjectFile("analytics/macros/bounded_activity_graph.sql"); + + expect(sql).toContain("min(ranked.started_at) AS started_at"); + expect(sql).toContain("max(coalesce(ranked.ended_at, ranked.started_at + INTERVAL 12 HOUR)) AS ended_at"); + expect(sql).not.toContain("any(best.started_at) AS started_at"); + expect(sql).not.toContain("any(best.ended_at) AS ended_at"); + }); + it("aggregates activity sensor summary from the bounded sensor intermediary", () => { expect(existsSync(new URL("./activity_sensor_summary_rows.sql", import.meta.url))).toBe(true); const sql = readModel("activity_sensor_summary_rows");