diff --git a/AGENTS.md b/AGENTS.md index a5500d427d..9483ad4dbb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -71,6 +71,7 @@ Provider-agnostic fitness/health data pipeline. Syncs data from various provider - **Do not replay heavyweight historical backfills in tests**: One-off production backfills and historical migration bodies (for example ClickHouse `0013_metric_stream_location_point`) do not need broad integration-test or mutation-test coverage. ClickHouse-backed integration tests should create the current final schema/read models directly and then seed/sync the minimal fixture data they need. Do not call the tracked production migration runner from router/test setup when that runner can replay expensive backfills. - **Production code drives, tests adapt**: Never add hacks, indirection, or complexity to production code just to make tests easier. No lazy initialization to avoid connections in tests, no conditional logic gated on `NODE_ENV === 'test'`, no runtime feature flags for testability. Tests should use mocks, dependency injection, or module-level mocking (`vi.mock`) to work around production code — not the other way around. If production code needs restructuring to be testable, that restructuring should also improve the production design (e.g., proper DI). - **No exports just for testability**: Never export a function, class, or variable solely because a test needs to access it. Exports define the public API of a module — every export should serve a production consumer. If a test needs to verify internal behavior, test it through the public interface instead. Comments like "exported for testing" are a code smell — if it's worth exporting, it's worth exporting for production use too. +- **No test/optional-only branches**: Never write a code branch that is only reachable in tests. If a dependency is always present in production, model it as a required (non-optional) dependency rather than an optional one guarded with `if (!dep) return []` — that guard is dead in production and silently hides infra failures (e.g. returning empty data instead of erroring when ClickHouse is down). Make required deps non-optional (drop the `?`, make constructor/`createApp` params required) and have test helpers always provide a stub (e.g. `makeMockSensorStore`). If a precondition genuinely can fail at runtime, fail loudly with a specific error — never return a benign empty result. - **Shared test utilities**: When multiple unit or integration tests need to share mock setups, utility functions, or test data, extract them into a local `test-helpers.ts` file within the same directory. Do not export these helpers from the source file being tested or import them from another `*.test.ts` file. - **Colocated unit tests**: Unit test files live next to the source file they test, named `.test.ts`. Do not use `__tests__/` directories. For example, `src/db/tokens.ts` has its unit test at `src/db/tokens.test.ts`. Integration tests (`*.integration.test.ts`) can live wherever makes sense. - **Test files map 1:1 to source files**: A unit test file should test one source file. If a test file grows too large because it covers several responsibilities, split the production source into smaller SOLID modules/classes and give each source file its own focused colocated test file. Do not split tests by arbitrary scenario while leaving the source file monolithic. diff --git a/docs/metric-stream-postgres-retirement.md b/docs/metric-stream-postgres-retirement.md new file mode 100644 index 0000000000..bbf387ea33 --- /dev/null +++ b/docs/metric-stream-postgres-retirement.md @@ -0,0 +1,120 @@ +# Retiring the Postgres metric stream + +Tracks the work to remove `fitness.metric_stream` from Postgres. Metric-stream +samples now flow provider/mobile → Redpanda → ClickHouse sink + R2 archive; +Postgres is no longer a metric-stream source or sink. This plan migrates the +remaining readers off Postgres so the table (and the PeerDB mirror that fed it) +can be dropped. + +## Current state + +- **Writers:** off Postgres. `db/metric-stream-writer.ts` only publishes to + Redpanda (`publishRows`/`replaceRows`); the Postgres sink was removed. +- **`fitness.metric_stream` (Postgres):** frozen — receives no new rows. Readers + below are serving stale data for anything written after the cutover. +- **ClickHouse serving copy:** `postgres_fitness.metric_stream`, fed by the + Redpanda `metric-stream-clickhouse-sink` (non-IMU only). Source for + `analytics.deduped_sensor` and all downstream dbt sensor models. **Kept.** +- **PeerDB PG→CH metric_stream mirror:** legacy shadow-validation path. To be + retired (redundant with the Redpanda sink; impossible once the PG table is + gone). +- **R2 archive:** every channel (incl. `imu`) archived as + `metric-stream/v1/date=…/hour=…/*.jsonl.gz`. + +## Naming decision + +The ClickHouse table lives in `postgres_fitness` (a schema of genuine PeerDB +mirrors of `fitness.*`), but metric_stream is now Redpanda-fed, not a PG mirror. +Move just that table to its own home: + +- **New name: `metric_stream.events`** (dedicated schema; parallels topic + `metric-stream-v1`). Keeps the `_peerdb_is_deleted/_version/_synced_at` + ReplacingMergeTree columns the dbt models expect. + +## Readers to migrate (API-serving, read `fitness.metric_stream`) + +| Reader | Channels | New source | +| --- | --- | --- | +| `routers/heart-rate.ts` | `heart_rate` (scalar) | `metric_stream.events` | +| `repositories/health-kit-sync-repository.ts` (SpO2, skin temp → daily_metrics) | scalar | `metric_stream.events` | +| `repositories/provider-detail-repository.ts` | counts/range per provider | `metric_stream.events` | +| `repositories/inertial-measurement-unit-repository.ts` | `imu`/IMU vectors | R2 object-coverage (see P2) | + +IMU is excluded from ClickHouse by the sink (`channel !== "imu"`) and is ~97% of +the table. The IMU pages (web `/inertial-measurement-unit`, mobile +`app/inertial-measurement-unit.tsx`) are **debug/observability** ("are we +receiving + storing IMU"), so they don't need raw samples — only presence/volume +per day, served cheaply from R2 object listing. + +--- + +## P0 — Rename CH `postgres_fitness.metric_stream` → `metric_stream.events` + +Prerequisite for P1 (readers should land on the final name once). + +1. Bootstrap DDL (`src/db/clickhouse-metric-stream-bootstrap.ts`): create + `metric_stream.events` (same columns/engine); keep helper that returns the + statements. +2. `clickhouse-sink.ts`: insert target → `metric_stream.events`. Update the + narrow insert interface table literal + integration test. +3. dbt: update the source feeding `deduped_sensor.sql` (and any model/source + yml referencing `postgres_fitness.metric_stream`) to `metric_stream.events`. + Run `pnpm lint:analytics-policy` + `pnpm lint:analytics-sql`. +4. CH migration: `RENAME TABLE postgres_fitness.metric_stream TO + metric_stream.events` (metadata-only on ReplacingMergeTree). Coordinate with + sink cutover (deploy sink + DDL together; rename is fast). +5. Delete-scope SQL (`clickhouse-sink.ts` delete path) → new table. +6. Validate: integration test inserts a row, dbt `deduped_sensor` still builds, + dashboards (recovery/strain) unaffected. + +## P1 — Migrate scalar readers to ClickHouse ← starting now + +Each reader: failing test first (integration against real CH), then swap query. + +1. `routers/heart-rate.ts`: `heart_rate` 1-min downsample → query + `metric_stream.events` (dedup with `FINAL` / `_peerdb_is_deleted = 0`). +2. `repositories/health-kit-sync-repository.ts`: SpO2 + wrist-temp daily + aggregation reads → ClickHouse. (Writes to `daily_metrics` unchanged.) +3. `repositories/provider-detail-repository.ts`: per-provider counts/time-range + → ClickHouse. +4. Drop the `fitness.v_metric_stream` materialized view + any other PG + metric_stream views/indexes once no reader references them. +5. Dual-platform: no client query shape change expected (server returns same + payload); verify web + mobile pages still render. + +## P2 — IMU debug pages → R2 coverage + +The IMU repo's hot reads are debug-only. Replace raw PG reads with a cheap R2 +object-coverage signal (no full scan, no decompress, no dedup needed). + +1. New read path: list R2 objects under `metric-stream/v1/date=…` (Cloudflare R2 + S3 API or ClickHouse `s3()` listing of `_path`/`_size`), aggregate by day → + presence + bytes. "Receiving" = recent objects; "storing" = bytes/day. +2. Repoint `inertialMeasurementUnitRouter` (`getDailyHeatmap`, + `getCoverageTimeline`, `getSyncStatus`) at the R2-coverage source. +3. Delete `InertialMeasurementUnitRepository`'s `fitness.metric_stream` queries. +4. Dual-platform: update web `InertialMeasurementUnitPage.tsx` + mobile + `app/inertial-measurement-unit.tsx` only if response shape changes; keep it + stable to avoid client churn. + +## P3 — Drop Postgres metric_stream + cleanup + +Only after P0–P2 are merged and verified, and the historical backfill is complete. + +1. Retire PeerDB metric_stream CDC mirror (so CH is fed only by Redpanda). +2. Delete `scripts/backfill-metric-stream-to-redpanda.ts` + its test (one-time + PG→Redpanda migration; source is being dropped). +3. Drizzle migration: drop `fitness.metric_stream` (+ remaining views, indexes, + any leftover triggers). Remove from `schema.ts`. +4. Remove `MetricStreamSourceRow`/PG-shaped types that only described the PG + table; keep the Redpanda event types (`events.ts`). +5. Docs: update `metric-stream-redpanda-r2-runbook.md` + `schema.md` to reflect + the dropped table; regenerate schema diagrams. + +## Validation gates (every P) + +- `pnpm lint`, `pnpm test:unit`, relevant `*.integration.test.ts`, all-package + `tsc --noEmit`. +- Integration tests run against real Postgres + ClickHouse (`docker compose up -d + db clickhouse redis`); never mock the DB for SQL/query changes. +- Each PR small, TDD, dual-platform parity (web + mobile). diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index dd7fc3c7ba..1f31e1a59f 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -61,12 +61,12 @@ export interface CreateAppOptions { export function createApp( db: import("dofek/db").Database, - sensorStore?: import("./repositories/activity-repository.ts").ActivitySensorStore, + sensorStore: import("./repositories/activity-repository.ts").ActivitySensorStore, options: CreateAppOptions = {}, ): express.Express { initSentry(); const app = express(); - const limitedSensorStore = sensorStore ? new LimitedActivitySensorStore(sensorStore) : undefined; + const limitedSensorStore = new LimitedActivitySensorStore(sensorStore); // ── Health check (before ALL middleware and other routes) ── app.get("/healthz", (_req, res) => { @@ -82,7 +82,7 @@ export function createApp( function setupRoutes( app: express.Express, db: import("dofek/db").Database, - sensorStore: import("./repositories/activity-repository.ts").ActivitySensorStore | undefined, + sensorStore: import("./repositories/activity-repository.ts").ActivitySensorStore, options: CreateAppOptions, ) { // ── Compression + Cookies ── diff --git a/packages/server/src/repositories/heart-rate-repository.integration.test.ts b/packages/server/src/repositories/heart-rate-repository.integration.test.ts new file mode 100644 index 0000000000..3c1fe115d6 --- /dev/null +++ b/packages/server/src/repositories/heart-rate-repository.integration.test.ts @@ -0,0 +1,177 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createClickHouseClientFromEnv } from "../../../../src/db/clickhouse.ts"; +import { buildClickHouseBootstrapStatementsForNativeMetricStream } from "../../../../src/db/clickhouse-metric-stream-bootstrap.ts"; +import { HeartRateRepository, type MetricStreamClickHouseReader } from "./heart-rate-repository.ts"; + +const testUserId = "00000000-0000-0000-0000-0000000000a1"; + +interface SeedRow { + id: string; + recorded_at: string; + provider_id: string; + channel: string; + scalar: number; + _peerdb_is_deleted: 0 | 1; + _peerdb_version: number; +} + +const client = createClickHouseClientFromEnv(); + +const reader: MetricStreamClickHouseReader = { + async query(schema, query, params) { + if (!client.query) throw new Error("ClickHouse client must support query"); + const result = await client.query({ query, query_params: params, format: "JSONEachRow" }); + const rows = await result.json(); + return rows.map((row) => schema.parse(row)); + }, +}; + +async function seed(rows: SeedRow[]): Promise { + await client.insert?.({ + table: "postgres_fitness.metric_stream", + format: "JSONEachRow", + values: rows.map((row) => ({ + id: row.id, + user_id: testUserId, + recorded_at: row.recorded_at, + provider_id: row.provider_id, + channel: row.channel, + scalar: row.scalar, + _peerdb_is_deleted: row._peerdb_is_deleted, + _peerdb_synced_at: "2026-04-12 00:00:00.000", + _peerdb_version: row._peerdb_version, + })), + }); +} + +describe("HeartRateRepository (integration)", () => { + beforeAll(async () => { + for (const statement of buildClickHouseBootstrapStatementsForNativeMetricStream("")) { + await client.command({ query: statement }); + } + await client.command({ + query: "DELETE FROM postgres_fitness.metric_stream WHERE user_id = {userId:UUID}", + query_params: { userId: testUserId }, + }); + + const supersededId = randomUUID(); + await seed([ + // whoop_ble: two minutes of data + { + id: randomUUID(), + recorded_at: "2026-04-12 10:00:00.000", + provider_id: "whoop_ble", + channel: "heart_rate", + scalar: 72, + _peerdb_is_deleted: 0, + _peerdb_version: 0, + }, + { + id: randomUUID(), + recorded_at: "2026-04-12 10:01:00.000", + provider_id: "whoop_ble", + channel: "heart_rate", + scalar: 74, + _peerdb_is_deleted: 0, + _peerdb_version: 0, + }, + // apple_health: same minute as whoop — must stay a separate source (no priority collapse) + { + id: randomUUID(), + recorded_at: "2026-04-12 10:00:30.000", + provider_id: "apple_health", + channel: "heart_rate", + scalar: 70, + _peerdb_is_deleted: 0, + _peerdb_version: 0, + }, + // version dedup: same id/time, v1 supersedes v0 — FINAL must keep 99, not 50 + { + id: supersededId, + recorded_at: "2026-04-12 11:00:00.000", + provider_id: "whoop_ble", + channel: "heart_rate", + scalar: 50, + _peerdb_is_deleted: 0, + _peerdb_version: 0, + }, + { + id: supersededId, + recorded_at: "2026-04-12 11:00:00.000", + provider_id: "whoop_ble", + channel: "heart_rate", + scalar: 99, + _peerdb_is_deleted: 0, + _peerdb_version: 1, + }, + // excluded rows + { + id: randomUUID(), + recorded_at: "2026-04-12 12:00:00.000", + provider_id: "whoop_ble", + channel: "heart_rate", + scalar: 88, + _peerdb_is_deleted: 1, + _peerdb_version: 1, + }, // deleted + { + id: randomUUID(), + recorded_at: "2026-04-11 10:00:00.000", + provider_id: "whoop_ble", + channel: "heart_rate", + scalar: 60, + _peerdb_is_deleted: 0, + _peerdb_version: 0, + }, // other day + { + id: randomUUID(), + recorded_at: "2026-04-12 13:00:00.000", + provider_id: "whoop_ble", + channel: "heart_rate", + scalar: 0, + _peerdb_is_deleted: 0, + _peerdb_version: 0, + }, // zero + { + id: randomUUID(), + recorded_at: "2026-04-12 14:00:00.000", + provider_id: "whoop_ble", + channel: "power", + scalar: 200, + _peerdb_is_deleted: 0, + _peerdb_version: 0, + }, // other channel + ]); + }, 120_000); + + afterAll(async () => { + await client.command({ + query: "DELETE FROM postgres_fitness.metric_stream WHERE user_id = {userId:UUID}", + query_params: { userId: testUserId }, + }); + await client.close?.(); + }); + + it("returns per-source minute samples, version-deduped, with non-matching rows excluded", async () => { + const repo = new HeartRateRepository(reader, testUserId, "UTC"); + const result = await repo.dailyBySource("2026-04-12"); + + const byProvider = Object.fromEntries(result.map((series) => [series.providerId, series])); + + expect(Object.keys(byProvider).sort()).toEqual(["apple_health", "whoop_ble"]); + + // apple_health stays separate from whoop_ble at the same minute + expect(byProvider.apple_health.samples).toEqual([ + { time: "2026-04-12T10:00:00.000Z", heartRate: 70 }, + ]); + + // whoop_ble: 10:00, 10:01, and the version-deduped 11:00 (99, not 50); + // deleted / other-day / zero / other-channel rows excluded + expect(byProvider.whoop_ble.samples).toEqual([ + { time: "2026-04-12T10:00:00.000Z", heartRate: 72 }, + { time: "2026-04-12T10:01:00.000Z", heartRate: 74 }, + { time: "2026-04-12T11:00:00.000Z", heartRate: 99 }, + ]); + }); +}); diff --git a/packages/server/src/repositories/heart-rate-repository.test.ts b/packages/server/src/repositories/heart-rate-repository.test.ts new file mode 100644 index 0000000000..67d89205e0 --- /dev/null +++ b/packages/server/src/repositories/heart-rate-repository.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, it, vi } from "vitest"; +import { HeartRateRepository } from "./heart-rate-repository.ts"; + +describe("HeartRateRepository", () => { + it("groups per-minute samples by provider", async () => { + const query = vi.fn(async () => [ + { provider_id: "whoop_ble", recorded_at: "2026-04-12T10:00:00Z", heart_rate: 72 }, + { provider_id: "whoop_ble", recorded_at: "2026-04-12T10:01:00Z", heart_rate: 74 }, + { provider_id: "apple_health", recorded_at: "2026-04-12T10:00:00Z", heart_rate: 70 }, + ]); + + const repo = new HeartRateRepository({ query }, "user-1", "UTC"); + const result = await repo.dailyBySource("2026-04-12"); + + expect(result).toHaveLength(2); + const ble = result.find((series) => series.providerId === "whoop_ble"); + expect(ble?.samples).toEqual([ + { time: "2026-04-12T10:00:00Z", heartRate: 72 }, + { time: "2026-04-12T10:01:00Z", heartRate: 74 }, + ]); + expect(result.find((series) => series.providerId === "apple_health")?.samples).toHaveLength(1); + }); + + it("passes the user id, timezone, and date through to the query", async () => { + const query = vi.fn(async () => []); + const repo = new HeartRateRepository({ query }, "user-9", "America/Los_Angeles"); + + await repo.dailyBySource("2026-04-12"); + + expect(query).toHaveBeenCalledWith(expect.anything(), expect.any(String), { + userId: "user-9", + timezone: "America/Los_Angeles", + date: "2026-04-12", + }); + }); + + it("returns an empty array when there are no rows", async () => { + const query = vi.fn(async () => []); + const repo = new HeartRateRepository({ query }, "user-1", "UTC"); + expect(await repo.dailyBySource("2026-04-12")).toEqual([]); + }); +}); diff --git a/packages/server/src/repositories/heart-rate-repository.ts b/packages/server/src/repositories/heart-rate-repository.ts new file mode 100644 index 0000000000..279b2e78c1 --- /dev/null +++ b/packages/server/src/repositories/heart-rate-repository.ts @@ -0,0 +1,96 @@ +import { providerLabel } from "@dofek/providers/providers"; +import { z } from "zod"; +import { timestampStringSchema } from "../lib/typed-sql.ts"; + +/** + * Minimal ClickHouse read surface this repository needs. Satisfied by + * `ActivitySensorStore`, but narrowed so the repository depends only on the + * raw-query escape hatch (interface segregation). + */ +export interface MetricStreamClickHouseReader { + query( + schema: TSchema, + query: string, + params?: Record, + ): Promise[]>; +} + +const heartRateRowSchema = z.object({ + provider_id: z.string(), + recorded_at: timestampStringSchema, + heart_rate: z.coerce.number(), +}); + +export interface HeartRateSourceSeries { + providerId: string; + providerLabel: string; + samples: { time: string; heartRate: number }[]; +} + +export class HeartRateRepository { + readonly #clickHouse: MetricStreamClickHouseReader; + readonly #userId: string; + readonly #timezone: string; + + constructor(clickHouse: MetricStreamClickHouseReader, userId: string, timezone: string) { + this.#clickHouse = clickHouse; + this.#userId = userId; + this.#timezone = timezone; + } + + /** + * Per-minute heart rate samples for a single day, grouped by source. + * + * Reads the Redpanda-fed ClickHouse metric-stream mirror for + * channel='heart_rate', downsampled to 1-minute bins (avg per bin). Rows are + * version-deduplicated (`FINAL` + `_peerdb_is_deleted = 0`) but NOT collapsed + * by provider priority, so every source is returned for overlay/comparison. + */ + async dailyBySource(date: string): Promise { + const rows = await this.#clickHouse.query( + heartRateRowSchema, + `SELECT + provider_id, + -- minute buckets always have zero sub-seconds; hardcode .000 so the + -- ISO-8601 string matches the existing client contract regardless of + -- ClickHouse's formatDateTime second-precision behavior. + formatDateTime(minute_bucket, '%Y-%m-%dT%H:%i:%S.000Z') AS recorded_at, + toInt32(round(avg(scalar))) AS heart_rate + FROM ( + SELECT + provider_id, + toStartOfMinute(toDateTime(recorded_at)) AS minute_bucket, + scalar + FROM postgres_fitness.metric_stream FINAL + WHERE user_id = {userId:String} + AND channel = 'heart_rate' + AND _peerdb_is_deleted = 0 + AND scalar > 0 + -- Range over the raw recorded_at so the ORDER BY key can prune, + -- instead of wrapping it in toDate(). Bounds are the user's local + -- day expressed as UTC instants. + AND recorded_at >= toDateTime({date:Date}, {timezone:String}) + AND recorded_at < toDateTime({date:Date}, {timezone:String}) + INTERVAL 1 DAY + ) + GROUP BY provider_id, minute_bucket + ORDER BY provider_id, minute_bucket`, + { userId: this.#userId, timezone: this.#timezone, date }, + ); + + const byProvider = new Map(); + for (const row of rows) { + let samples = byProvider.get(row.provider_id); + if (!samples) { + samples = []; + byProvider.set(row.provider_id, samples); + } + samples.push({ time: row.recorded_at, heartRate: row.heart_rate }); + } + + return Array.from(byProvider.entries()).map(([providerId, samples]) => ({ + providerId, + providerLabel: providerLabel(providerId), + samples, + })); + } +} diff --git a/packages/server/src/routers/food.integration.test.ts b/packages/server/src/routers/food.integration.test.ts index 32f9f0b8d9..a37dd4c7c4 100644 --- a/packages/server/src/routers/food.integration.test.ts +++ b/packages/server/src/routers/food.integration.test.ts @@ -3,6 +3,7 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { setupTestDatabase, type TestContext } from "../../../../src/db/test-helpers.ts"; import { createSession } from "../auth/session.ts"; import { createApp } from "../index.ts"; +import { makeMockSensorStore } from "./test-helpers.ts"; describe("Food router", () => { const TEST_USER_ID = "00000000-0000-0000-0000-000000000001"; @@ -17,7 +18,7 @@ describe("Food router", () => { const session = await createSession(testCtx.db, TEST_USER_ID); sessionCookie = `session=${session.sessionId}`; - const app = createApp(testCtx.db); + const app = createApp(testCtx.db, makeMockSensorStore()); await new Promise((resolve) => { server = app.listen(0, () => { const addr = server.address(); diff --git a/packages/server/src/routers/health-kit-sync.integration.test.ts b/packages/server/src/routers/health-kit-sync.integration.test.ts index 931d3830dd..3a070901b2 100644 --- a/packages/server/src/routers/health-kit-sync.integration.test.ts +++ b/packages/server/src/routers/health-kit-sync.integration.test.ts @@ -8,6 +8,7 @@ import { import type { MetricStreamEventPublisher } from "../../../../src/metric-stream/redpanda-producer.ts"; import { createSession } from "../auth/session.ts"; import { createApp } from "../index.ts"; +import { makeMockSensorStore } from "./test-helpers.ts"; describe("HealthKit sync router", () => { let server: ReturnType; @@ -83,7 +84,7 @@ describe("HealthKit sync router", () => { }, }; - const app = createApp(testCtx.db, undefined, { metricStreamPublisher }); + const app = createApp(testCtx.db, makeMockSensorStore(), { metricStreamPublisher }); await new Promise((resolve) => { server = app.listen(0, () => { const addr = server.address(); diff --git a/packages/server/src/routers/heart-rate.test.ts b/packages/server/src/routers/heart-rate.test.ts index c14d4222ed..850fc896b2 100644 --- a/packages/server/src/routers/heart-rate.test.ts +++ b/packages/server/src/routers/heart-rate.test.ts @@ -1,20 +1,17 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; -import { z } from "zod"; -import { createTestCallerFactory } from "./test-helpers.ts"; - -// Mock executeWithSchema -const mockExecuteWithSchema = vi.fn(); -vi.mock("../lib/typed-sql.ts", () => ({ - executeWithSchema: (...args: unknown[]) => mockExecuteWithSchema(...args), - dateStringSchema: z.string(), - timestampStringSchema: z - .union([z.string(), z.date()]) - .transform((value) => (value instanceof Date ? value.toISOString() : value)), -})); +import type { ActivitySensorStore } from "../repositories/activity-repository.ts"; +import { createTestCallerFactory, makeMockSensorStore } from "./test-helpers.ts"; vi.mock("../trpc.ts", async () => { const { initTRPC } = await import("@trpc/server"); - const trpc = initTRPC.context<{ db: unknown; userId: string; timezone: string }>().create(); + const trpc = initTRPC + .context<{ + db: unknown; + sensorStore?: ActivitySensorStore; + userId: string; + timezone: string; + }>() + .create(); return { router: trpc.router, protectedProcedure: trpc.procedure, @@ -23,10 +20,6 @@ vi.mock("../trpc.ts", async () => { }; }); -vi.mock("../lib/date-window.ts", () => ({ - endDateSchema: z.string().optional(), -})); - vi.mock("@dofek/providers/providers", () => ({ providerLabel: (id: string) => { const labels: Record = { @@ -39,48 +32,32 @@ vi.mock("@dofek/providers/providers", () => ({ describe("heartRateRouter", () => { beforeEach(() => { - mockExecuteWithSchema.mockReset(); + vi.clearAllMocks(); }); it("exports a dailyBySource procedure", async () => { const { heartRateRouter } = await import("./heart-rate.ts"); - expect(heartRateRouter).toBeDefined(); expect(heartRateRouter._def.procedures.dailyBySource).toBeDefined(); }); it("dailyBySource returns samples grouped by provider", async () => { - mockExecuteWithSchema.mockResolvedValue([ - { provider_id: "whoop_ble", recorded_at: "2026-04-12T10:00:00Z", heart_rate: 72 }, - { provider_id: "whoop_ble", recorded_at: "2026-04-12T10:01:00Z", heart_rate: 74 }, - { provider_id: "apple_health", recorded_at: "2026-04-12T10:00:00Z", heart_rate: 70 }, - ]); - const { heartRateRouter } = await import("./heart-rate.ts"); const createCaller = createTestCallerFactory(heartRateRouter); - const caller = createCaller({ db: {}, userId: "user-1", timezone: "UTC" }); + const caller = createCaller({ + db: { execute: vi.fn() }, + sensorStore: makeMockSensorStore([ + { provider_id: "whoop_ble", recorded_at: "2026-04-12T10:00:00Z", heart_rate: 72 }, + { provider_id: "whoop_ble", recorded_at: "2026-04-12T10:01:00Z", heart_rate: 74 }, + { provider_id: "apple_health", recorded_at: "2026-04-12T10:00:00Z", heart_rate: 70 }, + ]), + userId: "user-1", + timezone: "UTC", + }); const result = await caller.dailyBySource({ date: "2026-04-12" }); expect(result).toHaveLength(2); - - const bleSeries = result.find((series) => series.providerId === "whoop_ble"); - expect(bleSeries).toBeDefined(); - expect(bleSeries?.samples).toHaveLength(2); - expect(bleSeries?.samples[0].heartRate).toBe(72); - - const appleSeries = result.find((series) => series.providerId === "apple_health"); - expect(appleSeries).toBeDefined(); - expect(appleSeries?.samples).toHaveLength(1); - }); - - it("dailyBySource returns empty array when no data", async () => { - mockExecuteWithSchema.mockResolvedValue([]); - - const { heartRateRouter } = await import("./heart-rate.ts"); - const createCaller = createTestCallerFactory(heartRateRouter); - const caller = createCaller({ db: {}, userId: "user-1", timezone: "UTC" }); - - const result = await caller.dailyBySource({ date: "2026-04-12" }); - expect(result).toEqual([]); + expect(result.find((series) => series.providerId === "whoop_ble")?.samples).toHaveLength(2); + expect(result.find((series) => series.providerId === "apple_health")?.samples).toHaveLength(1); }); }); diff --git a/packages/server/src/routers/heart-rate.ts b/packages/server/src/routers/heart-rate.ts index 0f379fdf86..6eca1ce8d9 100644 --- a/packages/server/src/routers/heart-rate.ts +++ b/packages/server/src/routers/heart-rate.ts @@ -1,70 +1,25 @@ -import { providerLabel } from "@dofek/providers/providers"; -import { sql } from "drizzle-orm"; import { z } from "zod"; -import { executeWithSchema, timestampStringSchema } from "../lib/typed-sql.ts"; +import { + HeartRateRepository, + type HeartRateSourceSeries, +} from "../repositories/heart-rate-repository.ts"; import { CacheTTL, cachedProtectedQuery, router } from "../trpc.ts"; const dateInputSchema = z.object({ date: z.string().regex(/^\d{4}-\d{2}-\d{2}$/, "Expected YYYY-MM-DD date"), }); -const metricStreamRowSchema = z.object({ - provider_id: z.string(), - recorded_at: timestampStringSchema, - heart_rate: z.coerce.number(), -}); - -export interface HeartRateSourceSeries { - providerId: string; - providerLabel: string; - samples: { time: string; heartRate: number }[]; -} +export type { HeartRateSourceSeries }; export const heartRateRouter = router({ /** - * Per-minute heart rate samples for a single day, grouped by source. - * - * Queries metric_stream for channel='heart_rate', downsampled to 1-minute - * bins (avg per bin) to keep payload size reasonable. Returns one series - * per provider so the client can overlay them for comparison. + * Per-minute heart rate samples for a single day, grouped by source, read + * from the Redpanda-fed ClickHouse metric-stream mirror. */ dailyBySource: cachedProtectedQuery(CacheTTL.MEDIUM) .input(dateInputSchema) .query(async ({ ctx, input }): Promise => { - const rows = await executeWithSchema( - ctx.db, - metricStreamRowSchema, - sql`SELECT - provider_id, - date_trunc('minute', recorded_at) AS recorded_at, - ROUND(AVG(scalar))::int AS heart_rate - FROM fitness.metric_stream - WHERE user_id = ${ctx.userId} - AND channel = 'heart_rate' - AND recorded_at >= (${input.date}::date::timestamp AT TIME ZONE ${ctx.timezone}) - AND recorded_at < ((${input.date}::date + interval '1 day')::timestamp AT TIME ZONE ${ctx.timezone}) - AND scalar > 0 - GROUP BY provider_id, date_trunc('minute', recorded_at) - ORDER BY provider_id, recorded_at`, - ); - - if (rows.length === 0) return []; - - // Group by provider - const byProvider = new Map(); - for (const row of rows) { - let samples = byProvider.get(row.provider_id); - if (!samples) { - samples = []; - byProvider.set(row.provider_id, samples); - } - samples.push({ time: row.recorded_at, heartRate: row.heart_rate }); - } - - return Array.from(byProvider.entries()).map(([providerId, samples]) => ({ - providerId, - providerLabel: providerLabel(providerId), - samples, - })); + const repo = new HeartRateRepository(ctx.sensorStore, ctx.userId, ctx.timezone); + return repo.dailyBySource(input.date); }), }); diff --git a/packages/server/src/routers/settings.integration.test.ts b/packages/server/src/routers/settings.integration.test.ts index 51c7da710d..ecd317dba2 100644 --- a/packages/server/src/routers/settings.integration.test.ts +++ b/packages/server/src/routers/settings.integration.test.ts @@ -3,6 +3,7 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { setupTestDatabase, type TestContext } from "../../../../src/db/test-helpers.ts"; import { createSession } from "../auth/session.ts"; import { createApp } from "../index.ts"; +import { makeMockSensorStore } from "./test-helpers.ts"; const SETTINGS_TEST_USER_ID = "00000000-0000-0000-0000-0000000000f1"; @@ -28,7 +29,7 @@ describe("Settings router", () => { const session = await createSession(testCtx.db, SETTINGS_TEST_USER_ID); sessionCookie = `session=${session.sessionId}`; - const app = createApp(testCtx.db); + const app = createApp(testCtx.db, makeMockSensorStore()); await new Promise((resolve) => { server = app.listen(0, () => { const addr = server.address(); diff --git a/packages/server/src/routers/whoop-auth.integration.test.ts b/packages/server/src/routers/whoop-auth.integration.test.ts index 9208d2ee5d..b45ad57cb5 100644 --- a/packages/server/src/routers/whoop-auth.integration.test.ts +++ b/packages/server/src/routers/whoop-auth.integration.test.ts @@ -7,6 +7,7 @@ import { setupTestDatabase, type TestContext } from "../../../../src/db/test-hel import { createSession } from "../auth/session.ts"; import { createApp } from "../index.ts"; import { getWhoopVerificationChallengeStore } from "../lib/whoop-verification-challenge-store.ts"; +import { makeMockSensorStore } from "./test-helpers.ts"; const TEST_USER_ID = "00000000-0000-0000-0000-000000000001"; @@ -52,7 +53,7 @@ describe("whoopAuth router", () => { const session = await createSession(testCtx.db, TEST_USER_ID); sessionCookie = `session=${session.sessionId}`; - const app = createApp(testCtx.db); + const app = createApp(testCtx.db, makeMockSensorStore()); await new Promise((resolve) => { server = app.listen(0, () => { const addr = server.address(); diff --git a/packages/server/src/trpc.ts b/packages/server/src/trpc.ts index 1fe745951b..98962b864e 100644 --- a/packages/server/src/trpc.ts +++ b/packages/server/src/trpc.ts @@ -20,7 +20,7 @@ import type { ActivitySensorStore } from "./repositories/activity-repository.ts" export interface Context { db: Database; - sensorStore?: ActivitySensorStore; + sensorStore: ActivitySensorStore; metricStreamPublisher?: MetricStreamEventPublisher; userId: string | null; /** IANA timezone from client (e.g. "America/Los_Angeles"). Falls back to "UTC". */