Retire Postgres metric stream (P1): heart-rate reads from ClickHouse

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).

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<void> {
+  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 },
+    ]);
+  });
+});

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([]);
+  });
+});