From 202059f5cf4895d39dd1ab663cb4ea0d2bc73870 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 17:05:52 -0700 Subject: [PATCH 01/11] Fix metric stream sink deletes and move ClickHouse table to ingest. The Redpanda sink was stuck retrying ALTER UPDATE on version keys, blocking HR data from reaching strain and activity analytics. Co-authored-by: Cursor --- .../read_models/activity_location_sample.sql | 20 +++--- .../models/read_models/provider_stats.sql | 8 +-- .../read_model_microbatch.sql.test.ts | 23 +++---- analytics/models/sources.yml | 17 +++-- .../models/staging/sensor_scalar_sample.sql | 30 ++++---- .../heart-rate-repository.integration.test.ts | 6 +- .../src/repositories/heart-rate-repository.ts | 4 +- ...ider-detail-repository.integration.test.ts | 6 +- .../provider-detail-repository.test.ts | 8 +-- .../provider-detail-repository.ts | 10 +-- .../clickhouse-integration-test-helpers.ts | 30 +++----- .../src/routers/provider-detail.test.ts | 2 +- .../catch-up-clickhouse-metric-stream.test.ts | 11 +-- scripts/catch-up-clickhouse-metric-stream.ts | 19 ++--- scripts/seed-review-clickhouse.test.ts | 4 +- scripts/seed-review-clickhouse.ts | 2 +- src/db/clickhouse-activity-summary.test.ts | 2 +- src/db/clickhouse-deduped-sensor.ts | 10 +-- src/db/clickhouse-metric-stream-bootstrap.ts | 40 ++++------- src/db/clickhouse-migrations.test.ts | 69 +++++++++---------- .../0034_move_metric_stream_to_ingest.ts | 59 ++++++++++++++++ src/db/clickhouse-migrations/custom-runs.ts | 39 +++++++---- src/db/clickhouse-migrations/registry.ts | 3 + .../clickhouse-migrations/statement-runner.ts | 4 +- src/db/clickhouse-read-models.ts | 25 +++---- src/db/clickhouse.test.ts | 29 ++++---- src/db/clickhouse.ts | 5 +- .../clickhouse-sink.integration.test.ts | 5 +- src/metric-stream/clickhouse-sink.test.ts | 41 +++++++---- src/metric-stream/clickhouse-sink.ts | 43 +++++++++--- src/metric-stream/clickhouse-table.ts | 34 +++++++++ src/providers/garmin/sync-step-plan.ts | 4 +- 32 files changed, 359 insertions(+), 253 deletions(-) create mode 100644 src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts create mode 100644 src/metric-stream/clickhouse-table.ts diff --git a/analytics/models/read_models/activity_location_sample.sql b/analytics/models/read_models/activity_location_sample.sql index 4fda15e49a..b81b51051e 100644 --- a/analytics/models/read_models/activity_location_sample.sql +++ b/analytics/models/read_models/activity_location_sample.sql @@ -29,21 +29,21 @@ WITH activity_members AS ( location_versions AS ( SELECT * - FROM {{ source('postgres_fitness', 'metric_stream_freshness') }} + FROM {{ source('ingest', 'metric_stream_freshness') }} WHERE channel = 'location' - AND (point IS NOT NULL OR _peerdb_is_deleted = 1) + AND (point IS NOT NULL OR is_deleted = 1) ), location_rows AS ( SELECT id, - argMax(activity_id, _peerdb_version) AS member_activity_id, - argMax(user_id, _peerdb_version) AS user_id, - argMax(recorded_at, _peerdb_version) AS recorded_at, - argMax(provider_id, _peerdb_version) AS provider_id, - argMax(point, _peerdb_version) AS point, - argMax(_peerdb_synced_at, _peerdb_version) AS _peerdb_synced_at, - argMax(_peerdb_is_deleted, _peerdb_version) AS is_deleted + argMax(activity_id, version) AS member_activity_id, + argMax(user_id, version) AS user_id, + argMax(recorded_at, version) AS recorded_at, + argMax(provider_id, version) AS provider_id, + argMax(point, version) AS point, + argMax(ingested_at, version) AS ingested_at, + argMax(is_deleted, version) AS is_deleted FROM location_versions GROUP BY id ), @@ -96,7 +96,7 @@ SELECT )) AS lng, toUInt64(toUnixTimestamp64Nano(now64(9))) AS refresh_version, location_rows.is_deleted AS is_deleted, - greatest(location_rows._peerdb_synced_at, activity_members.source_synced_at) AS source_refreshed_at, + greatest(location_rows.ingested_at, activity_members.source_synced_at) AS source_refreshed_at, source_refreshed_at AS refreshed_at FROM location_points AS location_rows INNER JOIN activity_members diff --git a/analytics/models/read_models/provider_stats.sql b/analytics/models/read_models/provider_stats.sql index 97388f52a2..1f3748dd6d 100644 --- a/analytics/models/read_models/provider_stats.sql +++ b/analytics/models/read_models/provider_stats.sql @@ -43,8 +43,8 @@ WITH current_providers AS ( SELECT DISTINCT user_id, provider_id - FROM {{ source('postgres_fitness', 'metric_stream') }} FINAL - WHERE _peerdb_is_deleted = 0 + FROM {{ source('ingest', 'metric_stream') }} FINAL + WHERE is_deleted = 0 UNION DISTINCT @@ -175,8 +175,8 @@ metric_stream_counts AS ( user_id, provider_id, count() AS count - FROM {{ source('postgres_fitness', 'metric_stream') }} FINAL - WHERE _peerdb_is_deleted = 0 + FROM {{ source('ingest', 'metric_stream') }} FINAL + WHERE is_deleted = 0 GROUP BY user_id, provider_id ), 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 1a07aa2bff..80bd109892 100644 --- a/analytics/models/read_models/read_model_microbatch.sql.test.ts +++ b/analytics/models/read_models/read_model_microbatch.sql.test.ts @@ -259,20 +259,15 @@ describe("production analytics read-model build", () => { expect(sourcesYaml).toContain("name: metric_stream_freshness"); expect(sourcesYaml).toContain("identifier: metric_stream"); - expect(sourcesYaml).toContain("event_time: _peerdb_synced_at"); - expect(sensorScalarSampleSql).toContain("event_time='_peerdb_synced_at'"); - expect(sensorScalarSampleSql).toContain( - "source('postgres_fitness', 'metric_stream_freshness')", - ); + expect(sourcesYaml).toContain("name: ingest"); + expect(sourcesYaml).toContain("event_time: ingested_at"); + expect(sensorScalarSampleSql).toContain("event_time='ingested_at'"); + expect(sensorScalarSampleSql).toContain("source('ingest', 'metric_stream_freshness')"); expect(dedupedSensorSql).toContain("event_time='refreshed_at'"); expect(dedupedSensorSql).toContain("max(samples._peerdb_synced_at) AS source_refreshed_at"); expect(dedupedSensorSql).toContain("source_refreshed_at AS refreshed_at"); - expect(activityLocationSampleSql).toContain( - "source('postgres_fitness', 'metric_stream_freshness')", - ); - expect(activityLocationSampleSql).not.toContain( - "source('postgres_fitness', 'metric_stream')", - ); + expect(activityLocationSampleSql).toContain("source('ingest', 'metric_stream_freshness')"); + expect(activityLocationSampleSql).not.toContain("source('postgres_fitness', 'metric_stream')"); }); it("materializes activity location membership as a microbatch intermediary", () => { @@ -282,11 +277,11 @@ describe("production analytics read-model build", () => { expect(sql).toContain("incremental_strategy='microbatch'"); expect(sql).toContain("event_time='refreshed_at'"); expect(sql).toContain("lookback=3"); - expect(sql).toContain("source('postgres_fitness', 'metric_stream_freshness')"); + expect(sql).toContain("source('ingest', 'metric_stream_freshness')"); expect(sql).toContain("ref('deduped_activity_members')"); expect(sql).not.toContain("source('analytics', 'v_activity_members')"); expect(sql).toContain("channel = 'location'"); - expect(sql).toContain("argMax(point, _peerdb_version) AS point"); + expect(sql).toContain("argMax(point, version) AS point"); expect(sql).toContain("toString(point) AS point_text"); expect(sql).toContain("startsWith(location_rows.point_text, '{')"); expect(sql).toContain("JSONExtract(location_rows.point_text, 'coordinates', 'Array(Float64)')[2]"); @@ -319,7 +314,7 @@ describe("production analytics read-model build", () => { ); expect(activitySensorSampleSql).toContain("source_refreshed_at AS refreshed_at"); expect(activityLocationSampleSql).toContain( - "greatest(location_rows._peerdb_synced_at, activity_members.source_synced_at) AS source_refreshed_at", + "greatest(location_rows.ingested_at, activity_members.source_synced_at) AS source_refreshed_at", ); expect(activityLocationSampleSql).toContain("source_refreshed_at AS refreshed_at"); expect(activityLocationSampleSql).not.toContain("now64(9) AS refreshed_at"); diff --git a/analytics/models/sources.yml b/analytics/models/sources.yml index 5cb5b8bc4a..79481abe33 100644 --- a/analytics/models/sources.yml +++ b/analytics/models/sources.yml @@ -6,6 +6,16 @@ sources: tables: - name: v_activity - name: v_activity_members + - name: ingest + schema: ingest + tables: + - name: metric_stream + config: + event_time: recorded_at + - name: metric_stream_freshness + identifier: metric_stream + config: + event_time: ingested_at - name: postgres_fitness schema: postgres_fitness tables: @@ -17,13 +27,6 @@ sources: - name: journal_entry - name: lab_panel - name: lab_result - - name: metric_stream - config: - event_time: recorded_at - - name: metric_stream_freshness - identifier: metric_stream - config: - event_time: _peerdb_synced_at - name: provider - name: provider_priority - name: sensor_device_priority diff --git a/analytics/models/staging/sensor_scalar_sample.sql b/analytics/models/staging/sensor_scalar_sample.sql index cd403deadc..853738eec9 100644 --- a/analytics/models/staging/sensor_scalar_sample.sql +++ b/analytics/models/staging/sensor_scalar_sample.sql @@ -2,7 +2,7 @@ materialized='incremental', incremental_strategy='microbatch', unique_key='id', - event_time='_peerdb_synced_at', + event_time='ingested_at', begin='2026-01-01', batch_size='day', lookback=3, @@ -17,8 +17,8 @@ WITH metric_stream_versions AS ( SELECT * - FROM {{ source('postgres_fitness', 'metric_stream_freshness') }} - WHERE (scalar IS NOT null OR _peerdb_is_deleted = 1) + FROM {{ source('ingest', 'metric_stream_freshness') }} + WHERE (scalar IS NOT null OR is_deleted = 1) AND channel IN ( 'heart_rate', 'power', @@ -41,15 +41,15 @@ WITH metric_stream_versions AS ( metric_stream_rows AS ( SELECT id, - argMax(user_id, _peerdb_version) AS user_id, - argMax(recorded_at, _peerdb_version) AS recorded_at, - argMax(channel, _peerdb_version) AS channel, - argMax(provider_id, _peerdb_version) AS provider_id, - argMax(device_id, _peerdb_version) AS device_id, - coalesce(argMax(scalar, _peerdb_version), 0) AS scalar, - argMax(_peerdb_synced_at, _peerdb_version) AS _peerdb_synced_at, - argMax(_peerdb_is_deleted, _peerdb_version) AS _peerdb_is_deleted, - max(_peerdb_version) AS source_peerdb_version + argMax(user_id, version) AS user_id, + argMax(recorded_at, version) AS recorded_at, + argMax(channel, version) AS channel, + argMax(provider_id, version) AS provider_id, + argMax(device_id, version) AS device_id, + coalesce(argMax(scalar, version), 0) AS scalar, + argMax(ingested_at, version) AS ingested_at, + argMax(is_deleted, version) AS is_deleted, + max(version) AS source_version FROM metric_stream_versions GROUP BY id ), @@ -113,9 +113,9 @@ SELECT active_sensor_provider_priority.priority, 1000 ) AS provider_priority, - metric_stream_rows._peerdb_synced_at AS _peerdb_synced_at, - metric_stream_rows._peerdb_is_deleted AS _peerdb_is_deleted, - metric_stream_rows.source_peerdb_version AS _peerdb_version + metric_stream_rows.ingested_at AS _peerdb_synced_at, + metric_stream_rows.is_deleted AS _peerdb_is_deleted, + metric_stream_rows.source_version AS _peerdb_version FROM metric_stream_rows LEFT JOIN active_sensor_provider_priority ON active_sensor_provider_priority.provider_id = metric_stream_rows.provider_id diff --git a/packages/server/src/repositories/heart-rate-repository.integration.test.ts b/packages/server/src/repositories/heart-rate-repository.integration.test.ts index 3c1fe115d6..672ab567d4 100644 --- a/packages/server/src/repositories/heart-rate-repository.integration.test.ts +++ b/packages/server/src/repositories/heart-rate-repository.integration.test.ts @@ -29,7 +29,7 @@ const reader: MetricStreamClickHouseReader = { async function seed(rows: SeedRow[]): Promise { await client.insert?.({ - table: "postgres_fitness.metric_stream", + table: "ingest.metric_stream", format: "JSONEachRow", values: rows.map((row) => ({ id: row.id, @@ -51,7 +51,7 @@ describe("HeartRateRepository (integration)", () => { await client.command({ query: statement }); } await client.command({ - query: "DELETE FROM postgres_fitness.metric_stream WHERE user_id = {userId:UUID}", + query: "DELETE FROM ingest.metric_stream WHERE user_id = {userId:UUID}", query_params: { userId: testUserId }, }); @@ -147,7 +147,7 @@ describe("HeartRateRepository (integration)", () => { afterAll(async () => { await client.command({ - query: "DELETE FROM postgres_fitness.metric_stream WHERE user_id = {userId:UUID}", + query: "DELETE FROM ingest.metric_stream WHERE user_id = {userId:UUID}", query_params: { userId: testUserId }, }); await client.close?.(); diff --git a/packages/server/src/repositories/heart-rate-repository.ts b/packages/server/src/repositories/heart-rate-repository.ts index 279b2e78c1..401768638b 100644 --- a/packages/server/src/repositories/heart-rate-repository.ts +++ b/packages/server/src/repositories/heart-rate-repository.ts @@ -61,10 +61,10 @@ export class HeartRateRepository { provider_id, toStartOfMinute(toDateTime(recorded_at)) AS minute_bucket, scalar - FROM postgres_fitness.metric_stream FINAL + FROM ingest.metric_stream FINAL WHERE user_id = {userId:String} AND channel = 'heart_rate' - AND _peerdb_is_deleted = 0 + AND 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 diff --git a/packages/server/src/repositories/provider-detail-repository.integration.test.ts b/packages/server/src/repositories/provider-detail-repository.integration.test.ts index dfe658e318..0b1469e935 100644 --- a/packages/server/src/repositories/provider-detail-repository.integration.test.ts +++ b/packages/server/src/repositories/provider-detail-repository.integration.test.ts @@ -40,7 +40,7 @@ const clickHouse: BodyClickHouseStore = { async function seed(rows: SeedRow[]): Promise { await client.insert?.({ - table: "postgres_fitness.metric_stream", + table: "ingest.metric_stream", format: "JSONEachRow", values: rows.map((row) => ({ id: row.id, @@ -72,7 +72,7 @@ describe("ProviderDetailRepository metric stream (integration)", () => { await client.command({ query: statement }); } await client.command({ - query: "DELETE FROM postgres_fitness.metric_stream WHERE user_id = {userId:UUID}", + query: "DELETE FROM ingest.metric_stream WHERE user_id = {userId:UUID}", query_params: { userId: testUserId }, }); await seed([ @@ -137,7 +137,7 @@ describe("ProviderDetailRepository metric stream (integration)", () => { afterAll(async () => { await client.command({ - query: "DELETE FROM postgres_fitness.metric_stream WHERE user_id = {userId:UUID}", + query: "DELETE FROM ingest.metric_stream WHERE user_id = {userId:UUID}", query_params: { userId: testUserId }, }); await client.close?.(); diff --git a/packages/server/src/repositories/provider-detail-repository.test.ts b/packages/server/src/repositories/provider-detail-repository.test.ts index 3f9130e048..78026df852 100644 --- a/packages/server/src/repositories/provider-detail-repository.test.ts +++ b/packages/server/src/repositories/provider-detail-repository.test.ts @@ -19,7 +19,7 @@ describe("tableInfo", () => { ["bodyMeasurements", "analytics.v_body_measurement", "recorded_at", "id"], ["foodEntries", "fitness.food_entry", "date", "id"], ["healthEvents", "fitness.health_event", "start_date", "id"], - ["metricStream", "postgres_fitness.metric_stream", "recorded_at", "id"], + ["metricStream", "ingest.metric_stream", "recorded_at", "id"], ["nutritionDaily", "fitness.v_nutrition_daily", "date", "date"], ["labPanels", "fitness.lab_panel", "recorded_at", "id"], ["labResults", "fitness.lab_result", "recorded_at", "id"], @@ -226,8 +226,8 @@ describe("ProviderDetailRepository", () => { expect(query).toHaveBeenCalledTimes(1); // Reads the deduped Redpanda-fed mirror, not Postgres. - expect(query.mock.calls[0]?.[1]).toContain("postgres_fitness.metric_stream FINAL"); - expect(query.mock.calls[0]?.[1]).toContain("_peerdb_is_deleted = 0"); + expect(query.mock.calls[0]?.[1]).toContain("ingest.metric_stream FINAL"); + expect(query.mock.calls[0]?.[1]).toContain("is_deleted = 0"); // recorded_at is rendered in UTC so the literal 'Z' suffix is accurate. expect(query.mock.calls[0]?.[1]).toContain("'%Y-%m-%dT%H:%i:%S.%fZ', 'UTC'"); expect(query.mock.calls[0]?.[2]).toStrictEqual({ @@ -480,7 +480,7 @@ describe("ProviderDetailRepository", () => { idColumn: "id", }); expect(tableInfo("metricStream")).toStrictEqual({ - table: "postgres_fitness.metric_stream", + table: "ingest.metric_stream", orderColumn: "recorded_at", idColumn: "id", }); diff --git a/packages/server/src/repositories/provider-detail-repository.ts b/packages/server/src/repositories/provider-detail-repository.ts index 4e5505ecba..277d07a5f5 100644 --- a/packages/server/src/repositories/provider-detail-repository.ts +++ b/packages/server/src/repositories/provider-detail-repository.ts @@ -45,7 +45,7 @@ export function tableInfo(dataType: DataType): { return { table: "fitness.health_event", orderColumn: "start_date", idColumn: "id" }; case "metricStream": return { - table: "postgres_fitness.metric_stream", + table: "ingest.metric_stream", orderColumn: "recorded_at", idColumn: "id", }; @@ -231,7 +231,7 @@ export class ProviderDetailRepository { readonly #userId: string; // ClickHouse query runner. Serves the read-models that have moved off // Postgres: body measurements (analytics.v_body_measurement) and the - // Redpanda-fed metric stream (postgres_fitness.metric_stream). + // Redpanda-fed metric stream (ingest.metric_stream). readonly #clickHouse: BodyClickHouseStore | undefined; constructor( @@ -335,7 +335,7 @@ export class ProviderDetailRepository { /** * Raw metric-stream rows for a provider, read from the Redpanda-fed ClickHouse * mirror (Postgres `fitness.metric_stream` is retired). Version-deduplicated - * (`FINAL` + `_peerdb_is_deleted = 0`); recorded_at formatted to ISO-8601 Z. + * (`FINAL` + `is_deleted = 0`); recorded_at formatted to ISO-8601 Z. */ async #queryMetricStreamRecords( providerId: string, @@ -363,10 +363,10 @@ export class ProviderDetailRepository { channel, activity_id, scalar - FROM postgres_fitness.metric_stream FINAL + FROM ingest.metric_stream FINAL WHERE user_id = {userId:UUID} AND provider_id = {providerId:String} - AND _peerdb_is_deleted = 0 + AND is_deleted = 0 ${recordFilter} ORDER BY recorded_at DESC LIMIT {limit:UInt32} diff --git a/packages/server/src/routers/clickhouse-integration-test-helpers.ts b/packages/server/src/routers/clickhouse-integration-test-helpers.ts index 8ab62560d2..ab482dd559 100644 --- a/packages/server/src/routers/clickhouse-integration-test-helpers.ts +++ b/packages/server/src/routers/clickhouse-integration-test-helpers.ts @@ -41,6 +41,7 @@ export interface ClickHouseMetricStreamSeedRow { interface IsolatedClickHouseDatabases { analytics: string; + ingest: string; postgresFitness: string; } @@ -63,21 +64,6 @@ const clickHouseTestSetupSlotCount = 2; const clickHouseTestSetupSlotTimeoutMilliseconds = 55_000; const clickHouseTestSetupStaleSlotMilliseconds = 300_000; const rawTableSyncs: RawTableSync[] = [ - { - tableName: "metric_stream", - columns: [ - "id", - "activity_id", - "user_id", - "recorded_at", - "channel", - "provider_id", - "external_id", - "device_id", - "source_type", - "scalar", - ], - }, { tableName: "activity", columns: [ @@ -355,6 +341,7 @@ function rewriteClickHouseDatabaseNames( ): string { return query .replace(/\bpostgres_fitness\b/g, databases.postgresFitness) + .replace(/\bingest\b/g, databases.ingest) .replace(/\banalytics\b/g, databases.analytics); } @@ -1563,6 +1550,7 @@ export async function createClickHouseTestActivitySensorStore( const suffix = randomBytes(6).toString("hex"); const databases = { analytics: `analytics_test_${suffix}`, + ingest: `ingest_test_${suffix}`, postgresFitness: `postgres_fitness_test_${suffix}`, }; const rawClient = createClickHouseClientFromEnv(); @@ -1573,6 +1561,7 @@ export async function createClickHouseTestActivitySensorStore( testContext.addCleanup(async () => { handlesByContext.delete(testContext); await rawClient.command({ query: `DROP DATABASE IF EXISTS ${databases.analytics} SYNC` }); + await rawClient.command({ query: `DROP DATABASE IF EXISTS ${databases.ingest} SYNC` }); await rawClient.command({ query: `DROP DATABASE IF EXISTS ${databases.postgresFitness} SYNC` }); await rawClient.close?.(); }); @@ -1711,9 +1700,6 @@ async function syncClickHouseTestActivitySensorStoreWithClient( connectionString: string, ): Promise { for (const rawTableSync of rawTableSyncs) { - if (rawTableSync.tableName === "metric_stream") { - continue; - } await client.command({ query: `TRUNCATE TABLE postgres_fitness.${rawTableSync.tableName}`, }); @@ -1789,7 +1775,7 @@ export async function insertClickHouseMetricStreamRows( } await handle.setupClient.command({ - query: `INSERT INTO postgres_fitness.metric_stream ( + query: `INSERT INTO ingest.metric_stream ( id, activity_id, user_id, @@ -1803,9 +1789,9 @@ export async function insertClickHouseMetricStreamRows( vector, point, metadata, - _peerdb_synced_at, - _peerdb_is_deleted, - _peerdb_version + ingested_at, + is_deleted, + version ) VALUES ${rows.map(formatClickHouseMetricStreamSeedValue).join(",\n")}`, }); } diff --git a/packages/server/src/routers/provider-detail.test.ts b/packages/server/src/routers/provider-detail.test.ts index 73b4691af0..7ae09edc65 100644 --- a/packages/server/src/routers/provider-detail.test.ts +++ b/packages/server/src/routers/provider-detail.test.ts @@ -269,7 +269,7 @@ describe("providerDetailRouter", () => { ["bodyMeasurements", "analytics.v_body_measurement", "recorded_at", "id"], ["foodEntries", "fitness.food_entry", "date", "id"], ["healthEvents", "fitness.health_event", "start_date", "id"], - ["metricStream", "postgres_fitness.metric_stream", "recorded_at", "id"], + ["metricStream", "ingest.metric_stream", "recorded_at", "id"], ["nutritionDaily", "fitness.v_nutrition_daily", "date", "date"], ["labPanels", "fitness.lab_panel", "recorded_at", "id"], ["labResults", "fitness.lab_result", "recorded_at", "id"], diff --git a/scripts/catch-up-clickhouse-metric-stream.test.ts b/scripts/catch-up-clickhouse-metric-stream.test.ts index 3aa101d791..feac0322e5 100644 --- a/scripts/catch-up-clickhouse-metric-stream.test.ts +++ b/scripts/catch-up-clickhouse-metric-stream.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from "vitest"; +import { METRIC_STREAM_TABLE } from "../src/metric-stream/clickhouse-table.ts"; import { buildMetricStreamCatchUpStatement, buildMetricStreamCatchUpWindows, @@ -80,12 +81,12 @@ describe("buildMetricStreamCatchUpStatement", () => { start: new Date("2026-06-05T00:00:00Z"), }); - expect(sql).toContain("INSERT INTO postgres_fitness.metric_stream"); + expect(sql).toContain(`INSERT INTO ${METRIC_STREAM_TABLE}`); expect(sql).toContain( "postgresql('db:5432', 'health', 'metric_stream', 'health', 'p\\'ass', 'fitness')", ); expect(sql).toContain("LEFT ANY JOIN"); - expect(sql).toContain("FROM postgres_fitness.metric_stream FINAL"); + expect(sql).toContain(`FROM ${METRIC_STREAM_TABLE} FINAL`); expect(sql).toContain("metric_stream.channel != 'imu'"); expect(sql).toContain("existing_metric_stream.id IS NULL"); expect(sql).toContain( @@ -93,8 +94,8 @@ describe("buildMetricStreamCatchUpStatement", () => { ); expect(sql).toContain("toDateTime64('2026-06-05 00:00:00.000', 6, 'UTC')"); expect(sql).toContain("toDateTime64('2026-06-05 01:00:00.000', 6, 'UTC')"); - expect(sql).toContain("_peerdb_synced_at"); - expect(sql).toContain("_peerdb_is_deleted"); - expect(sql).toContain("_peerdb_version"); + expect(sql).toContain("ingested_at"); + expect(sql).toContain("is_deleted"); + expect(sql).toContain("version"); }); }); diff --git a/scripts/catch-up-clickhouse-metric-stream.ts b/scripts/catch-up-clickhouse-metric-stream.ts index c529261937..bc181f2ada 100644 --- a/scripts/catch-up-clickhouse-metric-stream.ts +++ b/scripts/catch-up-clickhouse-metric-stream.ts @@ -10,6 +10,7 @@ import { clickHouseStringLiteral, parsePostgresTimestamp, } from "../src/db/clickhouse-migrations/sql.ts"; +import { INGEST_DATABASE, METRIC_STREAM_TABLE } from "../src/metric-stream/clickhouse-table.ts"; export interface MetricStreamCatchUpOptions { start: Date; @@ -166,7 +167,7 @@ export function buildMetricStreamCatchUpStatement( options.postgresConnectionString, ); - return `INSERT INTO postgres_fitness.metric_stream ( + return `INSERT INTO ${METRIC_STREAM_TABLE} ( recorded_at, user_id, provider_id, @@ -178,9 +179,9 @@ export function buildMetricStreamCatchUpStatement( scalar, point, id, - _peerdb_synced_at, - _peerdb_is_deleted, - _peerdb_version + ingested_at, + is_deleted, + version ) SELECT metric_stream.recorded_at, @@ -222,9 +223,9 @@ SELECT ) ) AS point, metric_stream.id, - now64(9) AS _peerdb_synced_at, - 0 AS _peerdb_is_deleted, - toInt64(toUnixTimestamp64Milli(now64(3))) AS _peerdb_version + now64(9) AS ingested_at, + 0 AS is_deleted, + toInt64(toUnixTimestamp64Milli(now64(3))) AS version FROM ( SELECT metric_stream.recorded_at, @@ -245,7 +246,7 @@ FROM ( ) AS metric_stream LEFT ANY JOIN ( SELECT CAST(id, 'Nullable(UUID)') AS id - FROM postgres_fitness.metric_stream FINAL + FROM ${METRIC_STREAM_TABLE} FINAL WHERE recorded_at >= ${lowerBound} AND recorded_at < ${upperBound} ) AS existing_metric_stream @@ -267,7 +268,7 @@ async function runMetricStreamCatchUp( options: MetricStreamCatchUpOptions, ): Promise { const windows = buildMetricStreamCatchUpWindows(options); - await waitForClickHouseTable(client, "postgres_fitness", "metric_stream"); + await waitForClickHouseTable(client, INGEST_DATABASE, "metric_stream"); for (const window of windows) { console.log( diff --git a/scripts/seed-review-clickhouse.test.ts b/scripts/seed-review-clickhouse.test.ts index eeea374e41..2a4aa7855e 100644 --- a/scripts/seed-review-clickhouse.test.ts +++ b/scripts/seed-review-clickhouse.test.ts @@ -20,9 +20,9 @@ describe("seed-review-clickhouse", () => { ); expect(statements).toContain("TRUNCATE TABLE IF EXISTS postgres_fitness.activity"); - expect(statements).toContain("TRUNCATE TABLE IF EXISTS postgres_fitness.metric_stream"); + expect(statements).toContain("TRUNCATE TABLE IF EXISTS ingest.metric_stream"); expect(statements.join("\n")).toContain("INSERT INTO postgres_fitness.activity"); - expect(statements.join("\n")).toContain("INSERT INTO postgres_fitness.metric_stream"); + expect(statements.join("\n")).toContain("INSERT INTO ingest.metric_stream"); expect(statements.join("\n")).toContain( "WHERE user_id = '00000000-0000-0000-0000-000000000001'", ); diff --git a/scripts/seed-review-clickhouse.ts b/scripts/seed-review-clickhouse.ts index 164d280200..d22b9e1272 100644 --- a/scripts/seed-review-clickhouse.ts +++ b/scripts/seed-review-clickhouse.ts @@ -151,7 +151,7 @@ const reviewRawTableCopies: readonly ReviewRawTableCopy[] = [ whereClause: `user_id = ${clickHouseStringLiteral(USER_ID)}`, }, { - targetTable: "postgres_fitness.metric_stream", + targetTable: "ingest.metric_stream", sourceTable: "metric_stream", columns: [ "recorded_at", diff --git a/src/db/clickhouse-activity-summary.test.ts b/src/db/clickhouse-activity-summary.test.ts index e5e04710f0..27366f5174 100644 --- a/src/db/clickhouse-activity-summary.test.ts +++ b/src/db/clickhouse-activity-summary.test.ts @@ -23,7 +23,7 @@ describe("ClickHouse activity summary read model", () => { expect(sql).toContain("FROM analytics.activity_summary_rows FINAL"); expect(sql).toContain("WHERE is_deleted = 0"); expect(sql).not.toContain("JOIN analytics.deduped_sensor"); - expect(sql).not.toContain("FROM postgres_fitness.metric_stream"); + expect(sql).not.toContain("FROM ingest.metric_stream"); }); it("bootstrap statements expose both the target table and compatibility view", () => { diff --git a/src/db/clickhouse-deduped-sensor.ts b/src/db/clickhouse-deduped-sensor.ts index 2c88dda805..684acb4d0b 100644 --- a/src/db/clickhouse-deduped-sensor.ts +++ b/src/db/clickhouse-deduped-sensor.ts @@ -1,3 +1,5 @@ +import { METRIC_STREAM_TABLE } from "../metric-stream/clickhouse-table.ts"; + const sensorScalarChannels = [ "heart_rate", "power", @@ -82,9 +84,9 @@ SELECT metric_stream_rows.device_id AS device_id, assumeNotNull(metric_stream_rows.scalar) AS scalar, coalesce(device_priority_match.priority, active_sensor_provider_priority.priority, 1000) AS provider_priority, - metric_stream_rows._peerdb_synced_at AS _peerdb_synced_at, - metric_stream_rows._peerdb_is_deleted AS _peerdb_is_deleted, - metric_stream_rows._peerdb_version AS _peerdb_version + metric_stream_rows.ingested_at AS _peerdb_synced_at, + metric_stream_rows.is_deleted AS _peerdb_is_deleted, + metric_stream_rows.version AS _peerdb_version FROM metric_stream_rows LEFT JOIN active_sensor_provider_priority ON active_sensor_provider_priority.provider_id = metric_stream_rows.provider_id @@ -127,7 +129,7 @@ export function buildSensorScalarSampleBackfillSql(recordedAtRange?: RecordedAtR _peerdb_is_deleted, _peerdb_version ) -${sensorScalarSampleSelectSql("postgres_fitness.metric_stream", " FINAL", recordedAtRange)}`; +${sensorScalarSampleSelectSql(METRIC_STREAM_TABLE, " FINAL", recordedAtRange)}`; } function buildDedupedSensorTableSql(): string { diff --git a/src/db/clickhouse-metric-stream-bootstrap.ts b/src/db/clickhouse-metric-stream-bootstrap.ts index 48f9ac9cdd..69016225f3 100644 --- a/src/db/clickhouse-metric-stream-bootstrap.ts +++ b/src/db/clickhouse-metric-stream-bootstrap.ts @@ -8,36 +8,20 @@ import { } from "./clickhouse-read-models.ts"; import { buildRestingHeartRateSleepWindowTableSql } from "./clickhouse-resting-heart-rate.ts"; import { - peerDbMetadataColumnAlterStatements, - peerDbMetadataColumnDefinitions, - replacingMergeTreeTable, - standardViewHeader, -} from "./clickhouse-sql-helpers.ts"; + buildIngestMetricStreamCreateTableSql, + INGEST_DATABASE, + METRIC_STREAM_TABLE, +} from "../metric-stream/clickhouse-table.ts"; +import { standardViewHeader } from "./clickhouse-sql-helpers.ts"; export function buildClickHouseBootstrapStatementsForNativeMetricStream( postgresConnectionString: string, ): string[] { void postgresConnectionString; const metricStreamStatements = [ + `CREATE DATABASE IF NOT EXISTS ${INGEST_DATABASE}`, + buildIngestMetricStreamCreateTableSql(), "CREATE DATABASE IF NOT EXISTS postgres_fitness", - `CREATE TABLE IF NOT EXISTS postgres_fitness.metric_stream ( - id UUID, - activity_id Nullable(UUID), - user_id UUID, - recorded_at DateTime64(6, 'UTC'), - channel String, - provider_id String, - external_id Nullable(String), - device_id Nullable(String), - source_type Nullable(String), - scalar Nullable(Float32), - vector Array(Float32), - point String, - metadata String, -${peerDbMetadataColumnDefinitions} -) -${replacingMergeTreeTable("(user_id, activity_id, channel, recorded_at, id)")}`, - ...peerDbMetadataColumnAlterStatements("postgres_fitness.metric_stream"), ...buildPostgresFitnessRawTableStatements(), ...buildBodyMeasurementSampleProjectionStatements(), ]; @@ -83,11 +67,11 @@ linked_best_source AS ( ) AS row_number FROM ( SELECT * - FROM postgres_fitness.metric_stream FINAL + FROM ${METRIC_STREAM_TABLE} FINAL ) AS metric_stream INNER JOIN activity_members ON metric_stream.activity_id = activity_members.member_activity_id - WHERE metric_stream._peerdb_is_deleted = 0 + WHERE metric_stream.is_deleted = 0 AND metric_stream.channel = 'location' AND metric_stream.point != '' GROUP BY activity_members.activity_id, metric_stream.provider_id @@ -102,13 +86,13 @@ parsed_points AS ( metric_stream.recorded_at, metric_stream.provider_id, metric_stream.channel, - metric_stream._peerdb_is_deleted, + metric_stream.is_deleted, ( JSONExtract(metric_stream.point, 'coordinates', 'Array(Float64)')[1], JSONExtract(metric_stream.point, 'coordinates', 'Array(Float64)')[2] )::Point AS point - FROM (SELECT * FROM postgres_fitness.metric_stream FINAL) AS metric_stream - WHERE metric_stream._peerdb_is_deleted = 0 + FROM (SELECT * FROM ${METRIC_STREAM_TABLE} FINAL) AS metric_stream + WHERE metric_stream.is_deleted = 0 AND metric_stream.channel = 'location' AND metric_stream.point != '' ) diff --git a/src/db/clickhouse-migrations.test.ts b/src/db/clickhouse-migrations.test.ts index 51b9791215..ff70a5c01a 100644 --- a/src/db/clickhouse-migrations.test.ts +++ b/src/db/clickhouse-migrations.test.ts @@ -72,6 +72,7 @@ describe("buildClickHouseMigrationStatements", () => { "0031_activity_user_soft_delete.ts", "0032_deduped_activities_absent_source_links.ts", "0033_recreate_deduped_activities_column_order.ts", + "0034_move_metric_stream_to_ingest.ts", ]); }); @@ -262,24 +263,20 @@ SETTINGS allow_nullable_key = 1`); expect(sql).toContain("DROP TABLE IF EXISTS analytics.deduped_sensor"); expect(sql).toContain("DROP DATABASE IF EXISTS postgres_fitness SYNC"); expect(sql.match(/DROP DATABASE IF EXISTS postgres_fitness SYNC/g)).toHaveLength(1); - expect(sql).toContain("CREATE TABLE IF NOT EXISTS postgres_fitness.metric_stream"); - expect(sql.match(/CREATE TABLE IF NOT EXISTS postgres_fitness.metric_stream/g)).toHaveLength(4); - expect(sql).toContain( + expect(sql).toContain("CREATE TABLE IF NOT EXISTS ingest.metric_stream"); + expect(sql.match(/CREATE TABLE IF NOT EXISTS ingest\.metric_stream/g)).toHaveLength(5); + expect(sql).toContain("CREATE DATABASE IF NOT EXISTS ingest"); + expect(sql).toContain("DROP TABLE IF EXISTS postgres_fitness.metric_stream"); + expect(sql).not.toContain( "ALTER TABLE postgres_fitness.metric_stream ADD COLUMN IF NOT EXISTS _peerdb_synced_at DateTime64(9) DEFAULT now()", ); - expect(sql).toContain( - "ALTER TABLE postgres_fitness.metric_stream ADD COLUMN IF NOT EXISTS _peerdb_is_deleted Int8 DEFAULT 0", - ); - expect(sql).toContain( - "ALTER TABLE postgres_fitness.metric_stream ADD COLUMN IF NOT EXISTS _peerdb_version Int64 DEFAULT 0", - ); - expect(sql.indexOf("ADD COLUMN IF NOT EXISTS _peerdb_synced_at")).toBeLessThan( - sql.indexOf("FROM postgres_fitness.metric_stream"), + expect(sql.indexOf("CREATE TABLE IF NOT EXISTS ingest.metric_stream")).toBeLessThan( + sql.indexOf("FROM ingest.metric_stream"), ); expect(sql).toContain("point String"); - expect(sql).toContain("ENGINE = ReplacingMergeTree(_peerdb_version)"); - expect(sql).toContain("FROM postgres_fitness.metric_stream FINAL"); - expect(sql).toContain("WHERE _peerdb_is_deleted = 0"); + expect(sql).toContain("ENGINE = ReplacingMergeTree(version)"); + expect(sql).toContain("FROM ingest.metric_stream FINAL"); + expect(sql).toContain("WHERE is_deleted = 0"); expect(sql).toContain("ENGINE = ReplacingMergeTree"); expect(sql).toContain("CREATE TABLE IF NOT EXISTS analytics.sensor_scalar_sample"); expect(sql).toContain("CREATE TABLE IF NOT EXISTS analytics.deduped_sensor"); @@ -471,7 +468,7 @@ describe("runClickHouseMigrations", () => { const count = await runClickHouseMigrations(client, "postgres://health:fixture@db:5432/health"); - expect(count).toBe(33); + expect(count).toBe(34); expect(command).toHaveBeenCalledWith({ query: "CREATE DATABASE IF NOT EXISTS fitness" }); expect(command).toHaveBeenCalledWith({ query: "CREATE DATABASE IF NOT EXISTS analytics" }); expect(command).toHaveBeenCalledWith( @@ -491,7 +488,7 @@ describe("runClickHouseMigrations", () => { ); expect(command).toHaveBeenCalledWith( expect.objectContaining({ - query: expect.stringContaining("CREATE TABLE IF NOT EXISTS postgres_fitness.metric_stream"), + query: expect.stringContaining("CREATE TABLE IF NOT EXISTS ingest.metric_stream"), }), ); expect(command).not.toHaveBeenCalledWith( @@ -642,7 +639,7 @@ describe("runClickHouseMigrations", () => { ); expect(command).toHaveBeenCalledWith( expect.objectContaining({ - query: expect.stringContaining("INSERT INTO postgres_fitness.metric_stream"), + query: expect.stringContaining("INSERT INTO ingest.metric_stream"), }), ); expect(command).toHaveBeenCalledWith( @@ -654,7 +651,7 @@ describe("runClickHouseMigrations", () => { ); const backfillStatements = command.mock.calls .map(([options]) => String(options.query)) - .filter((queryText) => queryText.includes("INSERT INTO postgres_fitness.metric_stream")); + .filter((queryText) => queryText.includes("INSERT INTO ingest.metric_stream")); expect(backfillStatements).toHaveLength(24); expect(backfillStatements[0]).toContain( "metric_stream.recorded_at >= toDateTime64('2026-04-22 00:00:00.000', 6, 'UTC')", @@ -747,7 +744,7 @@ describe("runClickHouseMigrations", () => { expect(count).toBe(2); const backfillStatements = command.mock.calls .map(([options]) => String(options.query)) - .filter((queryText) => queryText.includes("INSERT INTO postgres_fitness.metric_stream")); + .filter((queryText) => queryText.includes("INSERT INTO ingest.metric_stream")); expect(backfillStatements).toHaveLength(12); expect(command).toHaveBeenCalledWith({ query: @@ -808,7 +805,7 @@ describe("runClickHouseMigrations", () => { ); expect(command).toHaveBeenCalledWith( expect.objectContaining({ - query: expect.stringContaining("INSERT INTO postgres_fitness.metric_stream"), + query: expect.stringContaining("INSERT INTO ingest.metric_stream"), }), ); }); @@ -844,7 +841,7 @@ describe("runClickHouseMigrations", () => { ); expect(command).not.toHaveBeenCalledWith( expect.objectContaining({ - query: expect.stringContaining("INSERT INTO postgres_fitness.metric_stream"), + query: expect.stringContaining("INSERT INTO ingest.metric_stream"), }), ); expect(command).toHaveBeenCalledWith({ @@ -904,17 +901,17 @@ describe("runClickHouseMigrations", () => { ); expect(command).toHaveBeenCalledWith( expect.objectContaining({ - query: "DROP TABLE IF EXISTS postgres_fitness.metric_stream", + query: "DROP TABLE IF EXISTS ingest.metric_stream", }), ); expect(command).toHaveBeenCalledWith( expect.objectContaining({ - query: expect.stringContaining("CREATE TABLE IF NOT EXISTS postgres_fitness.metric_stream"), + query: expect.stringContaining("CREATE TABLE IF NOT EXISTS ingest.metric_stream"), }), ); expect(command).toHaveBeenCalledWith( expect.objectContaining({ - query: expect.stringContaining("INSERT INTO postgres_fitness.metric_stream"), + query: expect.stringContaining("INSERT INTO ingest.metric_stream"), }), ); }); @@ -972,12 +969,12 @@ describe("runClickHouseMigrations", () => { ); expect(command).not.toHaveBeenCalledWith( expect.objectContaining({ - query: "DROP TABLE IF EXISTS postgres_fitness.metric_stream", + query: "DROP TABLE IF EXISTS ingest.metric_stream", }), ); expect(command).toHaveBeenCalledWith( expect.objectContaining({ - query: expect.stringContaining("INSERT INTO postgres_fitness.metric_stream"), + query: expect.stringContaining("INSERT INTO ingest.metric_stream"), }), ); }); @@ -1033,7 +1030,7 @@ describe("runClickHouseMigrations", () => { const backfillStatements = command.mock.calls .map(([options]) => String(options.query)) - .filter((queryText) => queryText.includes("INSERT INTO postgres_fitness.metric_stream")); + .filter((queryText) => queryText.includes("INSERT INTO ingest.metric_stream")); expect(backfillStatements).toHaveLength(6); expect(backfillStatements[0]).toContain( "metric_stream.recorded_at >= toDateTime64('2026-04-22 03:15:00.000', 6, 'UTC')", @@ -1085,7 +1082,7 @@ describe("runClickHouseMigrations", () => { const backfillStatements = command.mock.calls .map(([options]) => String(options.query)) - .filter((queryText) => queryText.includes("INSERT INTO postgres_fitness.metric_stream")); + .filter((queryText) => queryText.includes("INSERT INTO ingest.metric_stream")); expect(backfillStatements).toHaveLength(30); expect(backfillStatements[0]).toContain( "metric_stream.recorded_at >= toDateTime64('2026-04-22 00:00:00.000', 6, 'UTC')", @@ -1185,12 +1182,12 @@ describe("runClickHouseMigrations", () => { ); expect(command).toHaveBeenCalledWith( expect.objectContaining({ - query: "DROP TABLE IF EXISTS postgres_fitness.metric_stream", + query: "DROP TABLE IF EXISTS ingest.metric_stream", }), ); expect(command).toHaveBeenCalledWith( expect.objectContaining({ - query: expect.stringContaining("CREATE TABLE IF NOT EXISTS postgres_fitness.metric_stream"), + query: expect.stringContaining("CREATE TABLE IF NOT EXISTS ingest.metric_stream"), }), ); expect(command).toHaveBeenCalledWith( @@ -1200,7 +1197,7 @@ describe("runClickHouseMigrations", () => { ); expect(command).toHaveBeenCalledWith( expect.objectContaining({ - query: expect.stringContaining("INSERT INTO postgres_fitness.metric_stream"), + query: expect.stringContaining("INSERT INTO ingest.metric_stream"), }), ); }); @@ -1277,7 +1274,7 @@ describe("runClickHouseMigrations", () => { const backfillStatements = command.mock.calls .map(([options]) => String(options.query)) - .filter((queryText) => queryText.includes("INSERT INTO postgres_fitness.metric_stream")); + .filter((queryText) => queryText.includes("INSERT INTO ingest.metric_stream")); expect(backfillStatements).toHaveLength(288); expect(backfillStatements[0]).toContain( "metric_stream.recorded_at >= toDateTime64('2021-04-29 00:00:00.000', 6, 'UTC')", @@ -1353,7 +1350,7 @@ describe("runClickHouseMigrations", () => { expect(count).toBe(1); const backfillStatements = command.mock.calls .map(([options]) => String(options.query)) - .filter((queryText) => queryText.includes("INSERT INTO postgres_fitness.metric_stream")); + .filter((queryText) => queryText.includes("INSERT INTO ingest.metric_stream")); expect(backfillStatements).toHaveLength(12); expect(backfillStatements[0]).toContain( "metric_stream.recorded_at >= toDateTime64('2026-04-22 01:00:00.000', 6, 'UTC')", @@ -1396,7 +1393,7 @@ describe("runClickHouseMigrations", () => { ); expect(command).not.toHaveBeenCalledWith( expect.objectContaining({ - query: expect.stringContaining("INSERT INTO postgres_fitness.metric_stream"), + query: expect.stringContaining("INSERT INTO ingest.metric_stream"), }), ); }); @@ -1477,11 +1474,11 @@ describe("runClickHouseMigrationStatement", () => { { statement: "CREATE MATERIALIZED VIEW IF NOT EXISTS analytics.deduped_sensor TO analytics.deduped_sensor AS SELECT 1", - waitedTables: ["postgres_fitness.metric_stream", "analytics.v_activity_members"], + waitedTables: ["ingest.metric_stream", "analytics.v_activity_members"], }, { statement: "CREATE VIEW IF NOT EXISTS analytics.deduped_location AS SELECT 1", - waitedTables: ["postgres_fitness.metric_stream", "analytics.v_activity_members"], + waitedTables: ["ingest.metric_stream", "analytics.v_activity_members"], }, { statement: "CREATE VIEW IF NOT EXISTS analytics.activity_summary AS SELECT 1", diff --git a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts new file mode 100644 index 0000000000..2f77ee1197 --- /dev/null +++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts @@ -0,0 +1,59 @@ +import type { ClickHouseCommandClient } from "../clickhouse.ts"; +import { + buildIngestMetricStreamCreateTableSql, + INGEST_DATABASE, + LEGACY_METRIC_STREAM_TABLE, + METRIC_STREAM_TABLE, +} from "../../metric-stream/clickhouse-table.ts"; +import type { ClickHouseMigration } from "./types.ts"; + +async function legacyMetricStreamTableExists(client: ClickHouseCommandClient): Promise { + const result = await client.query({ + query: `SELECT count() AS count + FROM system.tables + WHERE database = 'postgres_fitness' + AND name = 'metric_stream'`, + format: "JSONEachRow", + }); + const rows = (await result.json()) as Array<{ count: string }>; + return Number(rows[0]?.count ?? 0) > 0; +} + +async function copyLegacyMetricStreamToIngest(client: ClickHouseCommandClient): Promise { + await client.command({ + query: `INSERT INTO ${METRIC_STREAM_TABLE} + SELECT + id, + activity_id, + user_id, + recorded_at, + channel, + provider_id, + external_id, + device_id, + source_type, + scalar, + vector, + point, + metadata, + _peerdb_synced_at AS ingested_at, + _peerdb_is_deleted AS is_deleted, + _peerdb_version AS version + FROM ${LEGACY_METRIC_STREAM_TABLE}`, + }); +} + +export function createMigration(): ClickHouseMigration { + return { + id: "0034_move_metric_stream_to_ingest", + statements: [`CREATE DATABASE IF NOT EXISTS ${INGEST_DATABASE}`, buildIngestMetricStreamCreateTableSql()], + run: async (client) => { + if (!(await legacyMetricStreamTableExists(client))) { + return; + } + + await copyLegacyMetricStreamToIngest(client); + await client.command({ query: `DROP TABLE IF EXISTS ${LEGACY_METRIC_STREAM_TABLE}` }); + }, + }; +} diff --git a/src/db/clickhouse-migrations/custom-runs.ts b/src/db/clickhouse-migrations/custom-runs.ts index d20ad4311d..b170c45d36 100644 --- a/src/db/clickhouse-migrations/custom-runs.ts +++ b/src/db/clickhouse-migrations/custom-runs.ts @@ -13,6 +13,10 @@ import { buildIncrementalDedupedSensorResetStatements, } from "../clickhouse-deduped-sensor.ts"; import { buildActivitySummaryReadModelStatements } from "../clickhouse-metric-stream-bootstrap.ts"; +import { + INGEST_DATABASE, + METRIC_STREAM_TABLE, +} from "../../metric-stream/clickhouse-table.ts"; import { buildPostgresFitnessActivityRawTableStatement } from "../clickhouse-raw-tables.ts"; import { buildActivityReadModelRefreshStatements } from "../clickhouse-read-models.ts"; import { @@ -192,7 +196,7 @@ export async function rebuildMetricStreamLocationPoint( ); await runClickHouseMigrationStatement( client, - "DROP TABLE IF EXISTS postgres_fitness.metric_stream", + `DROP TABLE IF EXISTS ${METRIC_STREAM_TABLE}`, ); } @@ -220,7 +224,7 @@ export async function replaceLegacyMetricStreamIfNeeded( "DROP TABLE IF EXISTS analytics.activity_summary", "DROP TABLE IF EXISTS analytics.deduped_sensor", "DROP TABLE IF EXISTS analytics.metric_stream_backfill_chunks", - "DROP TABLE IF EXISTS postgres_fitness.metric_stream", + `DROP TABLE IF EXISTS ${METRIC_STREAM_TABLE}`, ]; for (const statement of resetStatements) { @@ -270,17 +274,22 @@ async function metricStreamMirrorHasColumns( throw new Error("ClickHouse migrations require a query-capable client"); } const columnNames = columns.map(clickHouseStringLiteral).join(", "); - const result = await client.query({ - query: `SELECT count() AS migration_count + for (const database of [INGEST_DATABASE, "postgres_fitness"]) { + const result = await client.query({ + query: `SELECT count() AS migration_count FROM system.columns -WHERE database = 'postgres_fitness' +WHERE database = '${database}' AND table = 'metric_stream' AND name IN (${columnNames})`, - format: "JSONEachRow", - }); - const rows = await result.json(); - const parsedRows = z.array(migrationCountRowSchema).parse(rows); - return Number(parsedRows[0]?.migration_count ?? 0) === columns.length; + format: "JSONEachRow", + }); + const rows = await result.json(); + const parsedRows = z.array(migrationCountRowSchema).parse(rows); + if (Number(parsedRows[0]?.migration_count ?? 0) === columns.length) { + return true; + } + } + return false; } async function shouldReplacePostgresFitnessDatabase( @@ -307,7 +316,7 @@ async function shouldReplaceMetricStreamTable(client: ClickHouseCommandClient): } const result = await client.query({ query: - "SELECT engine FROM system.tables WHERE database = 'postgres_fitness' AND name = 'metric_stream'", + `SELECT engine FROM system.tables WHERE database = '${INGEST_DATABASE}' AND name = 'metric_stream'`, format: "JSONEachRow", }); const rows = await result.json(); @@ -341,8 +350,8 @@ async function backfillNativeMetricStream( client: ClickHouseCommandClient, postgresConnectionString: string, ): Promise { - logger.info("[migrate] Waiting for ClickHouse postgres_fitness.metric_stream table"); - await waitForClickHouseTable(client, "postgres_fitness", "metric_stream"); + logger.info(`[migrate] Waiting for ClickHouse ${METRIC_STREAM_TABLE} table`); + await waitForClickHouseTable(client, INGEST_DATABASE, "metric_stream"); const timescaleChunks = await fetchMetricStreamBackfillChunks(postgresConnectionString); const backfillRanges = timescaleChunks.flatMap(splitMetricStreamBackfillChunk); logger.info( @@ -526,7 +535,7 @@ function buildMetricStreamBackfillStatement( ): string { const lowerBound = clickHouseDateTimeLiteral(chunkStart); const upperBound = clickHouseDateTimeLiteral(chunkEnd); - return `INSERT INTO postgres_fitness.metric_stream ( + return `INSERT INTO ${METRIC_STREAM_TABLE} ( recorded_at, user_id, provider_id, @@ -582,7 +591,7 @@ SELECT FROM ${postgresMetricStreamSource} AS metric_stream LEFT JOIN ( SELECT CAST(id, 'Nullable(UUID)') AS id - FROM postgres_fitness.metric_stream + FROM ${METRIC_STREAM_TABLE} WHERE recorded_at >= ${lowerBound} AND recorded_at < ${upperBound} ) AS existing_metric_stream diff --git a/src/db/clickhouse-migrations/registry.ts b/src/db/clickhouse-migrations/registry.ts index 25575f21cd..5852f27bf6 100644 --- a/src/db/clickhouse-migrations/registry.ts +++ b/src/db/clickhouse-migrations/registry.ts @@ -31,6 +31,7 @@ import { createMigration as createMigration0030 } from "./0030_activity_mirror_o import { createMigration as createMigration0031 } from "./0031_activity_user_soft_delete.ts"; import { createMigration as createMigration0032 } from "./0032_deduped_activities_absent_source_links.ts"; import { createMigration as createMigration0033 } from "./0033_recreate_deduped_activities_column_order.ts"; +import { createMigration as createMigration0034 } from "./0034_move_metric_stream_to_ingest.ts"; import type { ClickHouseMigration, ClickHouseMigrationFactory } from "./types.ts"; const migrationFactories: ClickHouseMigrationFactory[] = [ @@ -67,6 +68,7 @@ const migrationFactories: ClickHouseMigrationFactory[] = [ createMigration0031, createMigration0032, createMigration0033, + createMigration0034, ]; export const clickHouseMigrationFileNames = [ @@ -103,6 +105,7 @@ export const clickHouseMigrationFileNames = [ "0031_activity_user_soft_delete.ts", "0032_deduped_activities_absent_source_links.ts", "0033_recreate_deduped_activities_column_order.ts", + "0034_move_metric_stream_to_ingest.ts", ] as const; export function clickHouseMigrations(postgresConnectionString: string): ClickHouseMigration[] { diff --git a/src/db/clickhouse-migrations/statement-runner.ts b/src/db/clickhouse-migrations/statement-runner.ts index 926b371165..bedc9e5b9d 100644 --- a/src/db/clickhouse-migrations/statement-runner.ts +++ b/src/db/clickhouse-migrations/statement-runner.ts @@ -9,11 +9,11 @@ export async function runClickHouseMigrationStatement( statement: string, ): Promise { if (statement.startsWith("CREATE MATERIALIZED VIEW IF NOT EXISTS analytics.deduped_sensor")) { - await waitForClickHouseTable(client, "postgres_fitness", "metric_stream"); + await waitForClickHouseTable(client, "ingest", "metric_stream"); await waitForClickHouseTable(client, "analytics", "v_activity_members"); } if (statement.startsWith("CREATE VIEW IF NOT EXISTS analytics.deduped_location")) { - await waitForClickHouseTable(client, "postgres_fitness", "metric_stream"); + await waitForClickHouseTable(client, "ingest", "metric_stream"); await waitForClickHouseTable(client, "analytics", "v_activity_members"); } if (statement.startsWith("CREATE VIEW IF NOT EXISTS analytics.activity_summary")) { diff --git a/src/db/clickhouse-read-models.ts b/src/db/clickhouse-read-models.ts index b1bd1edf75..900dd68f00 100644 --- a/src/db/clickhouse-read-models.ts +++ b/src/db/clickhouse-read-models.ts @@ -1,3 +1,4 @@ +import { METRIC_STREAM_TABLE } from "../metric-stream/clickhouse-table.ts"; import { buildProviderStatsTableSql } from "./clickhouse-provider-stats.ts"; import { peerDbMetadataColumnDefinitions, @@ -53,10 +54,10 @@ SELECT device_id, source_type, scalar, - _peerdb_synced_at, - _peerdb_is_deleted, - _peerdb_version -FROM postgres_fitness.metric_stream + ingested_at AS _peerdb_synced_at, + is_deleted AS _peerdb_is_deleted, + version AS _peerdb_version +FROM ${METRIC_STREAM_TABLE} WHERE channel IN ( ${bodyMeasurementChannelListSql()} )`; @@ -87,10 +88,10 @@ SELECT device_id, source_type, scalar, - _peerdb_synced_at, - _peerdb_is_deleted, - _peerdb_version -FROM postgres_fitness.metric_stream + ingested_at AS _peerdb_synced_at, + is_deleted AS _peerdb_is_deleted, + version AS _peerdb_version +FROM ${METRIC_STREAM_TABLE} WHERE channel IN ( ${bodyMeasurementChannelListSql()} )`; @@ -873,8 +874,8 @@ providers AS ( WHERE _peerdb_is_deleted = 0 UNION DISTINCT SELECT DISTINCT user_id, provider_id - FROM postgres_fitness.metric_stream FINAL - WHERE _peerdb_is_deleted = 0 + FROM ${METRIC_STREAM_TABLE} FINAL + WHERE is_deleted = 0 UNION DISTINCT SELECT DISTINCT user_id, provider_id FROM postgres_fitness.food_entry FINAL @@ -933,8 +934,8 @@ body_measurement_counts AS ( ), metric_stream_counts AS ( SELECT user_id, provider_id, count() AS count - FROM postgres_fitness.metric_stream FINAL - WHERE _peerdb_is_deleted = 0 + FROM ${METRIC_STREAM_TABLE} FINAL + WHERE is_deleted = 0 GROUP BY user_id, provider_id ), food_entry_counts AS ( diff --git a/src/db/clickhouse.test.ts b/src/db/clickhouse.test.ts index b7ce17332c..f5e0d99890 100644 --- a/src/db/clickhouse.test.ts +++ b/src/db/clickhouse.test.ts @@ -118,19 +118,14 @@ describe("buildClickHouseBootstrapStatements", () => { expect(sql).toContain("CREATE DATABASE IF NOT EXISTS analytics"); expect(sql).not.toContain("CREATE DATABASE IF NOT EXISTS fitness"); expect(sql).toContain("CREATE DATABASE IF NOT EXISTS postgres_fitness"); - expect(sql).toContain("CREATE TABLE IF NOT EXISTS postgres_fitness.metric_stream"); + expect(sql).toContain("CREATE DATABASE IF NOT EXISTS ingest"); + expect(sql).toContain("CREATE TABLE IF NOT EXISTS ingest.metric_stream"); expect(sql).toContain("vector Array(Float32)"); - expect(sql).toContain( - "ALTER TABLE postgres_fitness.metric_stream ADD COLUMN IF NOT EXISTS _peerdb_synced_at DateTime64(9) DEFAULT now()", - ); - expect(sql).toContain( - "ALTER TABLE postgres_fitness.metric_stream ADD COLUMN IF NOT EXISTS _peerdb_is_deleted Int8 DEFAULT 0", - ); - expect(sql).toContain( - "ALTER TABLE postgres_fitness.metric_stream ADD COLUMN IF NOT EXISTS _peerdb_version Int64 DEFAULT 0", - ); - expect(sql.indexOf("ADD COLUMN IF NOT EXISTS _peerdb_synced_at")).toBeLessThan( - sql.indexOf("FROM postgres_fitness.metric_stream"), + expect(sql).toContain("ingested_at DateTime64(9) DEFAULT now()"); + expect(sql).toContain("is_deleted Int8 DEFAULT 0"); + expect(sql).toContain("version Int64 DEFAULT 0"); + expect(sql).not.toContain( + "ALTER TABLE postgres_fitness.metric_stream ADD COLUMN IF NOT EXISTS _peerdb_synced_at", ); expect(sql).toContain("point String"); expect(sql).toContain("metadata String"); @@ -155,7 +150,7 @@ describe("buildClickHouseBootstrapStatements", () => { expect(sql).toContain("CREATE VIEW IF NOT EXISTS postgres_fitness.user_profile_current"); expect(sql).toContain("FROM postgres_fitness.user_profile FINAL"); expect(sql).toContain("WHERE _peerdb_is_deleted = 0"); - expect(sql).toContain("FROM postgres_fitness.metric_stream FINAL"); + expect(sql).toContain("FROM ingest.metric_stream FINAL"); expect(sql).toContain("ENGINE = ReplacingMergeTree"); expect(sql).not.toContain("ENGINE = MaterializedPostgreSQL"); expect(sql).not.toContain("materialized_postgresql_tables_list = 'metric_stream'"); @@ -184,7 +179,7 @@ describe("buildClickHouseBootstrapStatements", () => { expect(sql).not.toContain("SYSTEM WAIT VIEW analytics.activity_summary"); expect(sql).not.toContain("DROP TABLE IF EXISTS"); expect(sql).not.toContain("DROP VIEW IF EXISTS"); - expect(sql).toContain("FROM postgres_fitness.metric_stream"); + expect(sql).toContain("FROM ingest.metric_stream"); expect(sql).toContain("CREATE VIEW IF NOT EXISTS analytics.v_activity"); expect(sql).toContain("CREATE VIEW IF NOT EXISTS analytics.v_activity_members"); expect(sql).toContain("CREATE VIEW IF NOT EXISTS analytics.v_sleep"); @@ -235,7 +230,7 @@ describe("buildClickHouseBootstrapStatements", () => { expect(sql).not.toContain( "CREATE MATERIALIZED VIEW IF NOT EXISTS analytics.derived_resting_heart_rate", ); - expect(sql).toContain("FROM postgres_fitness.metric_stream FINAL"); + expect(sql).toContain("FROM ingest.metric_stream FINAL"); expect(sql).toContain("WHERE _peerdb_is_deleted = 0"); expect(sql).toContain("FROM analytics.v_activity"); expect(sql).toContain("FROM analytics.v_activity_members"); @@ -277,7 +272,7 @@ describe("bootstrapClickHouseFromEnv", () => { expect(command).not.toHaveBeenCalled(); expect(query).toHaveBeenCalledWith({ query: - "SELECT count() AS table_count FROM system.tables WHERE database = 'postgres_fitness' AND name = 'metric_stream'", + "SELECT count() AS table_count FROM system.tables WHERE database = 'ingest' AND name = 'metric_stream'", format: "JSONEachRow", }); expect(query).toHaveBeenCalledWith({ @@ -302,7 +297,7 @@ describe("bootstrapClickHouseFromEnv", () => { }); expect(query).toHaveBeenCalledWith({ query: - "SELECT count() AS column_count FROM system.columns WHERE database = 'postgres_fitness' AND table = 'metric_stream'", + "SELECT count() AS column_count FROM system.columns WHERE database = 'ingest' AND table = 'metric_stream'", format: "JSONEachRow", }); expect(query).toHaveBeenCalledWith({ diff --git a/src/db/clickhouse.ts b/src/db/clickhouse.ts index 0a816930ac..bb36c60885 100644 --- a/src/db/clickhouse.ts +++ b/src/db/clickhouse.ts @@ -1,5 +1,6 @@ import { createClient } from "@clickhouse/client"; import { buildClickHouseBootstrapStatementsForNativeMetricStream } from "./clickhouse-metric-stream-bootstrap.ts"; +import { METRIC_STREAM_TABLE } from "../metric-stream/clickhouse-table.ts"; export interface ClickHouseCommandClient { command(options: { @@ -111,12 +112,12 @@ export function createClickHouseClientFromEnv( } export async function bootstrapClickHouseFromEnv(client: ClickHouseCommandClient): Promise { - await waitForClickHouseTable(client, "postgres_fitness", "metric_stream"); + await waitForClickHouseTable(client, "ingest", "metric_stream"); await waitForClickHouseTable(client, "analytics", "deduped_sensor"); await waitForClickHouseTable(client, "analytics", "deduped_activities"); await waitForClickHouseTable(client, "analytics", "activity_summary"); await waitForClickHouseTable(client, "analytics", "activity_trend_daily"); - await smokeTestClickHouseTable(client, "postgres_fitness.metric_stream"); + await smokeTestClickHouseTable(client, METRIC_STREAM_TABLE); await smokeTestClickHouseTable(client, "analytics.deduped_sensor"); await smokeTestClickHouseTable(client, "analytics.deduped_activities"); await smokeTestClickHouseTable(client, "analytics.activity_summary"); diff --git a/src/metric-stream/clickhouse-sink.integration.test.ts b/src/metric-stream/clickhouse-sink.integration.test.ts index 7394cb114e..65173f9094 100644 --- a/src/metric-stream/clickhouse-sink.integration.test.ts +++ b/src/metric-stream/clickhouse-sink.integration.test.ts @@ -5,6 +5,7 @@ import { type ClickHouseMetricStreamInsertClient, insertMetricStreamEventsIntoClickHouse, } from "./clickhouse-sink.ts"; +import { METRIC_STREAM_TABLE } from "./clickhouse-table.ts"; import { createMetricStreamEvent } from "./events.ts"; const testUserId = "00000000-0000-0000-0000-000000000001"; @@ -20,7 +21,7 @@ function assertInsertCapable( async function removeTestEvent(client: ClickHouseClient): Promise { await client.command({ - query: "DELETE FROM postgres_fitness.metric_stream WHERE id = {id:UUID}", + query: `DELETE FROM ${METRIC_STREAM_TABLE} WHERE id = {id:UUID}`, query_params: { id: testEventId }, }); } @@ -58,7 +59,7 @@ describe("metric stream ClickHouse sink (integration)", () => { await insertMetricStreamEventsIntoClickHouse(client, [event]); const result = await client.query<{ count: string }>({ - query: "SELECT count() AS count FROM postgres_fitness.metric_stream WHERE id = {id:UUID}", + query: `SELECT count() AS count FROM ${METRIC_STREAM_TABLE} WHERE id = {id:UUID}`, query_params: { id: testEventId }, format: "JSONEachRow", }); diff --git a/src/metric-stream/clickhouse-sink.test.ts b/src/metric-stream/clickhouse-sink.test.ts index af4d81057d..111fc73bd7 100644 --- a/src/metric-stream/clickhouse-sink.test.ts +++ b/src/metric-stream/clickhouse-sink.test.ts @@ -1,4 +1,5 @@ import { afterEach, describe, expect, it, vi } from "vitest"; +import { METRIC_STREAM_TABLE } from "./clickhouse-table.ts"; import { applyMetricStreamEventsToClickHouse, insertMetricStreamEventsIntoClickHouse, @@ -38,14 +39,14 @@ afterEach(() => { }); describe("insertMetricStreamEventsIntoClickHouse", () => { - it("inserts non-IMU events into the existing analytics source table", async () => { + it("inserts non-IMU events into the ingest metric stream table", async () => { const insert = vi.fn(async () => undefined); const inserted = await insertMetricStreamEventsIntoClickHouse({ insert }, [heartRateEvent]); expect(inserted).toBe(1); expect(insert).toHaveBeenCalledWith({ - table: "postgres_fitness.metric_stream", + table: METRIC_STREAM_TABLE, format: "JSONEachRow", clickhouse_settings: { date_time_input_format: "best_effort" }, values: [ @@ -56,7 +57,7 @@ describe("insertMetricStreamEventsIntoClickHouse", () => { external_id: "hk:heart-rate-1", device_id: "Apple Watch", scalar: 72, - _peerdb_is_deleted: 0, + is_deleted: 0, }), ], }); @@ -94,7 +95,7 @@ describe("insertMetricStreamEventsIntoClickHouse", () => { expect(row.activity_id).toBe("20000000-0000-4000-8000-000000000001"); expect(row.scalar).toBe(72); expect(row.point).toBe('{"type":"Point","coordinates":[-122.4,37.8]}'); - expect(row._peerdb_version).toBe(0); + expect(row.version).toBe(0); }); it("normalizes EWKT point events into the GeoJSON string ClickHouse read models expect", () => { @@ -139,15 +140,18 @@ describe("applyMetricStreamEventsToClickHouse", () => { expect(applied).toBe(1); expect(command).toHaveBeenCalledWith({ - query: - "ALTER TABLE postgres_fitness.metric_stream UPDATE _peerdb_is_deleted = 1, _peerdb_version = {peerdb_version:Int64} WHERE toString(activity_id) = {activity_id:String}", + query: expect.stringContaining(`INSERT INTO ${METRIC_STREAM_TABLE}`), query_params: { - peerdb_version: expect.any(Number), + delete_version: expect.any(Number), activity_id: "20000000-0000-4000-8000-000000000001", }, }); + expect(String(command.mock.calls[0]?.[0].query)).toContain("1 AS is_deleted"); + expect(String(command.mock.calls[0]?.[0].query)).toContain( + "toString(activity_id) = {activity_id:String}", + ); expect(insert).toHaveBeenCalledWith({ - table: "postgres_fitness.metric_stream", + table: METRIC_STREAM_TABLE, values: [expect.objectContaining({ id: heartRateEvent.id })], format: "JSONEachRow", clickhouse_settings: { date_time_input_format: "best_effort" }, @@ -174,13 +178,13 @@ describe("applyMetricStreamEventsToClickHouse", () => { expect(applied).toBe(2); expect(insert).toHaveBeenCalledTimes(2); expect(insert).toHaveBeenNthCalledWith(1, { - table: "postgres_fitness.metric_stream", + table: METRIC_STREAM_TABLE, values: [expect.objectContaining({ id: heartRateEvent.id })], format: "JSONEachRow", clickhouse_settings: { date_time_input_format: "best_effort" }, }); expect(insert).toHaveBeenNthCalledWith(2, { - table: "postgres_fitness.metric_stream", + table: METRIC_STREAM_TABLE, values: [expect.objectContaining({ id: secondHeartRateEvent.id })], format: "JSONEachRow", clickhouse_settings: { date_time_input_format: "best_effort" }, @@ -188,7 +192,7 @@ describe("applyMetricStreamEventsToClickHouse", () => { expect(command).toHaveBeenCalledOnce(); }); - it("renders every supported replacement scope predicate into the ClickHouse mutation", async () => { + it("renders every supported replacement scope predicate into the tombstone insert", async () => { const command = vi.fn(async () => undefined); const insert = vi.fn(async () => undefined); @@ -206,10 +210,9 @@ describe("applyMetricStreamEventsToClickHouse", () => { expect(insert).not.toHaveBeenCalled(); expect(command).toHaveBeenCalledWith({ - query: - "ALTER TABLE postgres_fitness.metric_stream UPDATE _peerdb_is_deleted = 1, _peerdb_version = {peerdb_version:Int64} WHERE toString(user_id) = {user_id:String} AND provider_id = {provider_id:String} AND external_id = {external_id:String} AND channel = {channel:String} AND toString(activity_id) = {activity_id:String} AND recorded_at >= parseDateTime64BestEffort({recorded_at_start:String}) AND recorded_at < parseDateTime64BestEffort({recorded_at_end:String})", + query: expect.stringContaining(`INSERT INTO ${METRIC_STREAM_TABLE}`), query_params: { - peerdb_version: expect.any(Number), + delete_version: expect.any(Number), user_id: "10000000-0000-4000-8000-000000000001", provider_id: "fitbit", external_id: null, @@ -219,6 +222,16 @@ describe("applyMetricStreamEventsToClickHouse", () => { recorded_at_end: "2026-03-02T00:00:00.000Z", }, }); + const query = String(command.mock.calls[0]?.[0].query); + expect(query).toContain("toString(user_id) = {user_id:String}"); + expect(query).toContain("provider_id = {provider_id:String}"); + expect(query).toContain("external_id = {external_id:String}"); + expect(query).toContain("channel = {channel:String}"); + expect(query).toContain("toString(activity_id) = {activity_id:String}"); + expect(query).toContain( + "recorded_at >= parseDateTime64BestEffort({recorded_at_start:String})", + ); + expect(query).toContain("recorded_at < parseDateTime64BestEffort({recorded_at_end:String})"); }); it("rejects replacement delete scopes that produce no ClickHouse conditions", async () => { diff --git a/src/metric-stream/clickhouse-sink.ts b/src/metric-stream/clickhouse-sink.ts index 607433fb27..2f536f64e8 100644 --- a/src/metric-stream/clickhouse-sink.ts +++ b/src/metric-stream/clickhouse-sink.ts @@ -1,4 +1,5 @@ import { type ClickHouseClient, createClickHouseClientFromEnv } from "../db/clickhouse.ts"; +import { METRIC_STREAM_TABLE } from "./clickhouse-table.ts"; import { isMetricStreamDeletedEvent, type MetricStreamDeleteScope, @@ -13,7 +14,7 @@ import { export interface ClickHouseMetricStreamInsertClient { command?(options: { query: string; query_params?: Record }): Promise; insert(options: { - table: "postgres_fitness.metric_stream"; + table: typeof METRIC_STREAM_TABLE; values: readonly ClickHouseMetricStreamRow[]; format: "JSONEachRow"; clickhouse_settings?: Record; @@ -32,9 +33,9 @@ export interface ClickHouseMetricStreamRow { scalar: number | null; point: string | null; id: string; - _peerdb_synced_at: string; - _peerdb_is_deleted: 0; - _peerdb_version: number; + ingested_at: string; + is_deleted: 0 | 1; + version: number; } function isClickHouseReplicatedEvent(event: MetricStreamEventV1): boolean { @@ -70,9 +71,9 @@ export function mapMetricStreamEventToClickHouseRow( scalar: event.scalar ?? null, point: normalizePointForClickHouse(event.point), id: event.id, - _peerdb_synced_at: new Date().toISOString(), - _peerdb_is_deleted: 0, - _peerdb_version: 0, + ingested_at: new Date().toISOString(), + is_deleted: 0, + version: 0, }; } @@ -87,10 +88,10 @@ export async function insertMetricStreamEventsIntoClickHouse( } await client.insert({ - table: "postgres_fitness.metric_stream", + table: METRIC_STREAM_TABLE, values: rows, format: "JSONEachRow", - // recorded_at / _peerdb_synced_at are ISO-8601 strings with a trailing Z; + // recorded_at / ingested_at are ISO-8601 strings with a trailing Z; // ClickHouse only accepts that offset format with best_effort datetime parsing. clickhouse_settings: { date_time_input_format: "best_effort" }, }); @@ -145,11 +146,31 @@ export async function markMetricStreamScopeDeletedInClickHouse( throw new Error("ClickHouse metric-stream replacement requires a command-capable client"); } const queryParams: Record = { - peerdb_version: Date.now(), + delete_version: Date.now(), }; const conditions = clickHouseDeleteScopeConditions(scope, queryParams); await client.command({ - query: `ALTER TABLE postgres_fitness.metric_stream UPDATE _peerdb_is_deleted = 1, _peerdb_version = {peerdb_version:Int64} WHERE ${conditions.join(" AND ")}`, + query: `INSERT INTO ${METRIC_STREAM_TABLE} + SELECT + id, + activity_id, + user_id, + recorded_at, + channel, + provider_id, + external_id, + device_id, + source_type, + scalar, + vector, + point, + metadata, + now64(9) AS ingested_at, + 1 AS is_deleted, + greatest(version + 1, {delete_version:Int64}) AS version + FROM ${METRIC_STREAM_TABLE} FINAL + WHERE is_deleted = 0 + AND ${conditions.join(" AND ")}`, query_params: queryParams, }); } diff --git a/src/metric-stream/clickhouse-table.ts b/src/metric-stream/clickhouse-table.ts new file mode 100644 index 0000000000..acb03e934b --- /dev/null +++ b/src/metric-stream/clickhouse-table.ts @@ -0,0 +1,34 @@ +export const INGEST_DATABASE = "ingest"; +export const METRIC_STREAM_TABLE = `${INGEST_DATABASE}.metric_stream`; +export const LEGACY_METRIC_STREAM_TABLE = "postgres_fitness.metric_stream"; +export const METRIC_STREAM_ORDER_BY = "(user_id, activity_id, channel, recorded_at, id)"; + +export const metricStreamIngestMetadataColumnDefinitions = ` ingested_at DateTime64(9) DEFAULT now(), + is_deleted Int8 DEFAULT 0, + version Int64 DEFAULT 0`; + +export function metricStreamReplacingMergeTreeEngine(): string { + return `ENGINE = ReplacingMergeTree(version) +ORDER BY ${METRIC_STREAM_ORDER_BY} +SETTINGS allow_nullable_key = 1`; +} + +export function buildIngestMetricStreamCreateTableSql(): string { + return `CREATE TABLE IF NOT EXISTS ${METRIC_STREAM_TABLE} ( + id UUID, + activity_id Nullable(UUID), + user_id UUID, + recorded_at DateTime64(6, 'UTC'), + channel String, + provider_id String, + external_id Nullable(String), + device_id Nullable(String), + source_type Nullable(String), + scalar Nullable(Float32), + vector Array(Float32), + point String, + metadata String, +${metricStreamIngestMetadataColumnDefinitions} +) +${metricStreamReplacingMergeTreeEngine()}`; +} diff --git a/src/providers/garmin/sync-step-plan.ts b/src/providers/garmin/sync-step-plan.ts index a1b5844292..103689e991 100644 --- a/src/providers/garmin/sync-step-plan.ts +++ b/src/providers/garmin/sync-step-plan.ts @@ -144,13 +144,13 @@ async function listGarminDatesWithMetricStreamChannel( const result = await client.query({ query: ` SELECT DISTINCT toString(toDate(recorded_at)) AS date - FROM postgres_fitness.metric_stream FINAL + FROM ingest.metric_stream FINAL WHERE user_id = {userId:UUID} AND provider_id = {providerId:String} AND channel = {channel:String} AND recorded_at >= parseDateTime64BestEffort({rangeStart:String}) AND recorded_at < parseDateTime64BestEffort({rangeEnd:String}) - AND _peerdb_is_deleted = 0 + AND is_deleted = 0 `, format: "JSONEachRow", query_params: { From 3a5ff311210b71165ed6c073cf518d0e73c56746 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 17:07:44 -0700 Subject: [PATCH 02/11] Remove unused peerDbMetadataColumnAlterStatements export for Knip. The metric stream bootstrap no longer adds PeerDB metadata columns via ALTER TABLE. Co-authored-by: Cursor --- src/db/clickhouse-sql-helpers.ts | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/src/db/clickhouse-sql-helpers.ts b/src/db/clickhouse-sql-helpers.ts index 8f10667b24..9769d71744 100644 --- a/src/db/clickhouse-sql-helpers.ts +++ b/src/db/clickhouse-sql-helpers.ts @@ -2,18 +2,6 @@ export const peerDbMetadataColumnDefinitions = ` _peerdb_synced_at DateTime64(9 _peerdb_is_deleted Int8 DEFAULT 0, _peerdb_version Int64 DEFAULT 0`; -const peerDbMetadataColumns = [ - "_peerdb_synced_at DateTime64(9) DEFAULT now()", - "_peerdb_is_deleted Int8 DEFAULT 0", - "_peerdb_version Int64 DEFAULT 0", -] as const; - -export function peerDbMetadataColumnAlterStatements(tableName: string): string[] { - return peerDbMetadataColumns.map( - (metadataColumn) => `ALTER TABLE ${tableName} ADD COLUMN IF NOT EXISTS ${metadataColumn}`, - ); -} - export function replacingMergeTreeTable(orderBy: string): string { return `ENGINE = ReplacingMergeTree(_peerdb_version) ORDER BY ${orderBy} From 689bc9a0bf3e0d7ebdd19efc34457abdf43eacdb Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 17:10:24 -0700 Subject: [PATCH 03/11] Fix TypeScript errors in ingest migration and test helpers. Add the ingest database to isolated ClickHouse test bootstrap objects, guard optional query in migration 0034, and tighten sink tombstone test assertions. Co-authored-by: Cursor --- .../clickhouse-integration-test-helpers.ts | 56 ++++++------------- .../0034_move_metric_stream_to_ingest.ts | 3 + src/metric-stream/clickhouse-sink.test.ts | 16 +++++- 3 files changed, 32 insertions(+), 43 deletions(-) diff --git a/packages/server/src/routers/clickhouse-integration-test-helpers.ts b/packages/server/src/routers/clickhouse-integration-test-helpers.ts index ab482dd559..0a48ce79c9 100644 --- a/packages/server/src/routers/clickhouse-integration-test-helpers.ts +++ b/packages/server/src/routers/clickhouse-integration-test-helpers.ts @@ -1584,88 +1584,64 @@ async function bootstrapClickHouseTestSchema( client: ClickHouseClient, connectionString: string, ): Promise { + const defaultTestDatabases: IsolatedClickHouseDatabases = { + analytics: "analytics", + ingest: "ingest", + postgresFitness: "postgres_fitness", + }; + for (const statement of buildClickHouseBootstrapStatements(connectionString)) { await client.command({ query: statement }); } await client.command({ query: `CREATE VIEW IF NOT EXISTS analytics.deduped_activities AS -${buildTestDedupedActivitiesSelectSql({ - analytics: "analytics", - postgresFitness: "postgres_fitness", -})}`, +${buildTestDedupedActivitiesSelectSql(defaultTestDatabases)}`, }); await client.command({ query: `CREATE VIEW IF NOT EXISTS analytics.activity_sensor_sample AS -${buildTestActivitySensorSampleSelectSql({ - analytics: "analytics", - postgresFitness: "postgres_fitness", -})}`, +${buildTestActivitySensorSampleSelectSql(defaultTestDatabases)}`, }); await client.command({ query: `CREATE VIEW IF NOT EXISTS analytics.activity_location_sample AS -${buildTestActivityLocationSampleSelectSql({ - analytics: "analytics", - postgresFitness: "postgres_fitness", -})}`, +${buildTestActivityLocationSampleSelectSql(defaultTestDatabases)}`, }); await client.command({ query: `CREATE VIEW IF NOT EXISTS analytics.daily_recovery_inputs AS -${buildTestDailyRecoveryInputsSelectSql({ - analytics: "analytics", - postgresFitness: "postgres_fitness", -})}`, +${buildTestDailyRecoveryInputsSelectSql(defaultTestDatabases)}`, }); await client.command({ query: `CREATE VIEW IF NOT EXISTS analytics.daily_sleep AS -${buildTestDailySleepSelectSql({ - analytics: "analytics", - postgresFitness: "postgres_fitness", -})}`, +${buildTestDailySleepSelectSql(defaultTestDatabases)}`, }); await client.command({ query: `CREATE VIEW IF NOT EXISTS analytics.daily_recovery AS -${buildTestRecoveryReadModelSelectSql({ - analytics: "analytics", - postgresFitness: "postgres_fitness", -})}`, +${buildTestRecoveryReadModelSelectSql(defaultTestDatabases)}`, }); await client.command({ query: `CREATE VIEW IF NOT EXISTS analytics.daily_activity_load AS -${buildTestDailyActivityLoadSelectSql({ - analytics: "analytics", - postgresFitness: "postgres_fitness", -})}`, +${buildTestDailyActivityLoadSelectSql(defaultTestDatabases)}`, }); await client.command({ query: `CREATE VIEW IF NOT EXISTS analytics.daily_strain AS -${buildTestStrainReadModelSelectSql({ - analytics: "analytics", - postgresFitness: "postgres_fitness", -})}`, +${buildTestStrainReadModelSelectSql(defaultTestDatabases)}`, }); await client.command({ query: `CREATE VIEW IF NOT EXISTS analytics.healthspan_activity_zone_minutes AS -${buildTestHealthspanActivityZoneMinutesSelectSql({ - analytics: "analytics", - postgresFitness: "postgres_fitness", -})}`, +${buildTestHealthspanActivityZoneMinutesSelectSql(defaultTestDatabases)}`, }); await client.command({ query: `CREATE VIEW IF NOT EXISTS analytics.weekly_healthspan AS -${buildTestHealthspanReadModelSelectSql({ - analytics: "analytics", - postgresFitness: "postgres_fitness", -})}`, +${buildTestHealthspanReadModelSelectSql(defaultTestDatabases)}`, }); await client.command({ query: buildActivityVo2MaxEstimateTableSql() }); } diff --git a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts index 2f77ee1197..f5d34ada14 100644 --- a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts +++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts @@ -8,6 +8,9 @@ import { import type { ClickHouseMigration } from "./types.ts"; async function legacyMetricStreamTableExists(client: ClickHouseCommandClient): Promise { + if (!client.query) { + throw new Error("ClickHouse migrations require a query-capable client"); + } const result = await client.query({ query: `SELECT count() AS count FROM system.tables diff --git a/src/metric-stream/clickhouse-sink.test.ts b/src/metric-stream/clickhouse-sink.test.ts index 111fc73bd7..456222c2f2 100644 --- a/src/metric-stream/clickhouse-sink.test.ts +++ b/src/metric-stream/clickhouse-sink.test.ts @@ -32,6 +32,16 @@ const imuEvent = { channel: "imu", } satisfies MetricStreamEventV1; +function firstCommandQuery( + command: ReturnType, +): string { + const call = command.mock.calls[0]?.[0] as { query: string } | undefined; + if (!call) { + throw new Error("expected command call"); + } + return String(call.query); +} + afterEach(() => { vi.doUnmock("../db/clickhouse.ts"); vi.doUnmock("./redpanda-consumer.ts"); @@ -146,8 +156,8 @@ describe("applyMetricStreamEventsToClickHouse", () => { activity_id: "20000000-0000-4000-8000-000000000001", }, }); - expect(String(command.mock.calls[0]?.[0].query)).toContain("1 AS is_deleted"); - expect(String(command.mock.calls[0]?.[0].query)).toContain( + expect(String(firstCommandQuery(command))).toContain("1 AS is_deleted"); + expect(String(firstCommandQuery(command))).toContain( "toString(activity_id) = {activity_id:String}", ); expect(insert).toHaveBeenCalledWith({ @@ -222,7 +232,7 @@ describe("applyMetricStreamEventsToClickHouse", () => { recorded_at_end: "2026-03-02T00:00:00.000Z", }, }); - const query = String(command.mock.calls[0]?.[0].query); + const query = firstCommandQuery(command); expect(query).toContain("toString(user_id) = {user_id:String}"); expect(query).toContain("provider_id = {provider_id:String}"); expect(query).toContain("external_id = {external_id:String}"); From 55004af2eddbc7c16bab493a9baa13a478d60ebc Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 17:18:09 -0700 Subject: [PATCH 04/11] Fix CI lint and unit tests for ingest metric stream migration. Update test expectations and Biome formatting for the ingest database rename, and make migration 0034 tolerate legacy table count query mocks. Co-authored-by: Cursor --- ...lickhouse-integration-test-helpers.test.ts | 4 +-- src/db/clickhouse-metric-stream-bootstrap.ts | 10 +++---- src/db/clickhouse-migrations.test.ts | 1 - .../0034_move_metric_stream_to_ingest.ts | 30 ++++++++++++++++--- src/db/clickhouse-migrations/custom-runs.ts | 13 ++------ src/db/clickhouse.ts | 2 +- src/metric-stream/clickhouse-sink.test.ts | 16 ++++------ 7 files changed, 43 insertions(+), 33 deletions(-) diff --git a/packages/server/src/routers/clickhouse-integration-test-helpers.test.ts b/packages/server/src/routers/clickhouse-integration-test-helpers.test.ts index 9f3bd700b8..9bbeb8869d 100644 --- a/packages/server/src/routers/clickhouse-integration-test-helpers.test.ts +++ b/packages/server/src/routers/clickhouse-integration-test-helpers.test.ts @@ -104,7 +104,7 @@ describe("clickhouse integration test helpers", () => { expect( setupCommands.some( (command) => - command.includes("CREATE TABLE IF NOT EXISTS postgres_fitness_test_") && + command.includes("CREATE TABLE IF NOT EXISTS ingest_test_") && command.includes(".metric_stream"), ), ).toBe(true); @@ -300,7 +300,7 @@ describe("clickhouse integration test helpers", () => { (command) => command.includes("INSERT INTO analytics_test_") && command.includes(".sensor_scalar_sample") && - command.includes("FROM postgres_fitness_test_"), + command.includes("FROM ingest_test_"), ), ).toBe(true); expect( diff --git a/src/db/clickhouse-metric-stream-bootstrap.ts b/src/db/clickhouse-metric-stream-bootstrap.ts index 69016225f3..16717b67a4 100644 --- a/src/db/clickhouse-metric-stream-bootstrap.ts +++ b/src/db/clickhouse-metric-stream-bootstrap.ts @@ -1,3 +1,8 @@ +import { + buildIngestMetricStreamCreateTableSql, + INGEST_DATABASE, + METRIC_STREAM_TABLE, +} from "../metric-stream/clickhouse-table.ts"; import { buildIncrementalActivitySummaryStatements } from "./clickhouse-activity-summary.ts"; import { buildActivityTrendDailyCreateReadModelStatements } from "./clickhouse-activity-trend-read-model.ts"; import { buildIncrementalDedupedSensorStatements } from "./clickhouse-deduped-sensor.ts"; @@ -7,11 +12,6 @@ import { buildBodyMeasurementSampleProjectionStatements, } from "./clickhouse-read-models.ts"; import { buildRestingHeartRateSleepWindowTableSql } from "./clickhouse-resting-heart-rate.ts"; -import { - buildIngestMetricStreamCreateTableSql, - INGEST_DATABASE, - METRIC_STREAM_TABLE, -} from "../metric-stream/clickhouse-table.ts"; import { standardViewHeader } from "./clickhouse-sql-helpers.ts"; export function buildClickHouseBootstrapStatementsForNativeMetricStream( diff --git a/src/db/clickhouse-migrations.test.ts b/src/db/clickhouse-migrations.test.ts index ff70a5c01a..ec77e28660 100644 --- a/src/db/clickhouse-migrations.test.ts +++ b/src/db/clickhouse-migrations.test.ts @@ -266,7 +266,6 @@ SETTINGS allow_nullable_key = 1`); expect(sql).toContain("CREATE TABLE IF NOT EXISTS ingest.metric_stream"); expect(sql.match(/CREATE TABLE IF NOT EXISTS ingest\.metric_stream/g)).toHaveLength(5); expect(sql).toContain("CREATE DATABASE IF NOT EXISTS ingest"); - expect(sql).toContain("DROP TABLE IF EXISTS postgres_fitness.metric_stream"); expect(sql).not.toContain( "ALTER TABLE postgres_fitness.metric_stream ADD COLUMN IF NOT EXISTS _peerdb_synced_at DateTime64(9) DEFAULT now()", ); diff --git a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts index f5d34ada14..2d81e8fe30 100644 --- a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts +++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts @@ -1,12 +1,32 @@ -import type { ClickHouseCommandClient } from "../clickhouse.ts"; import { buildIngestMetricStreamCreateTableSql, INGEST_DATABASE, LEGACY_METRIC_STREAM_TABLE, METRIC_STREAM_TABLE, } from "../../metric-stream/clickhouse-table.ts"; +import type { ClickHouseCommandClient } from "../clickhouse.ts"; import type { ClickHouseMigration } from "./types.ts"; +function isLegacyTableCountRow(value: unknown): value is { count: string } { + return ( + typeof value === "object" && + value !== null && + "count" in value && + typeof value.count === "string" + ); +} + +function parseLegacyTableCount(rows: unknown): number { + if (!Array.isArray(rows) || rows.length === 0) { + return 0; + } + const first = rows[0]; + if (!isLegacyTableCountRow(first)) { + return 0; + } + return Number(first.count); +} + async function legacyMetricStreamTableExists(client: ClickHouseCommandClient): Promise { if (!client.query) { throw new Error("ClickHouse migrations require a query-capable client"); @@ -18,8 +38,7 @@ async function legacyMetricStreamTableExists(client: ClickHouseCommandClient): P AND name = 'metric_stream'`, format: "JSONEachRow", }); - const rows = (await result.json()) as Array<{ count: string }>; - return Number(rows[0]?.count ?? 0) > 0; + return parseLegacyTableCount(await result.json()) > 0; } async function copyLegacyMetricStreamToIngest(client: ClickHouseCommandClient): Promise { @@ -49,7 +68,10 @@ async function copyLegacyMetricStreamToIngest(client: ClickHouseCommandClient): export function createMigration(): ClickHouseMigration { return { id: "0034_move_metric_stream_to_ingest", - statements: [`CREATE DATABASE IF NOT EXISTS ${INGEST_DATABASE}`, buildIngestMetricStreamCreateTableSql()], + statements: [ + `CREATE DATABASE IF NOT EXISTS ${INGEST_DATABASE}`, + buildIngestMetricStreamCreateTableSql(), + ], run: async (client) => { if (!(await legacyMetricStreamTableExists(client))) { return; diff --git a/src/db/clickhouse-migrations/custom-runs.ts b/src/db/clickhouse-migrations/custom-runs.ts index b170c45d36..e583e0bb72 100644 --- a/src/db/clickhouse-migrations/custom-runs.ts +++ b/src/db/clickhouse-migrations/custom-runs.ts @@ -1,6 +1,7 @@ import { Client, escapeIdentifier } from "pg"; import { z } from "zod"; import { logger } from "../../logger.ts"; +import { INGEST_DATABASE, METRIC_STREAM_TABLE } from "../../metric-stream/clickhouse-table.ts"; import { buildClickHouseBootstrapStatements, type ClickHouseCommandClient, @@ -13,10 +14,6 @@ import { buildIncrementalDedupedSensorResetStatements, } from "../clickhouse-deduped-sensor.ts"; import { buildActivitySummaryReadModelStatements } from "../clickhouse-metric-stream-bootstrap.ts"; -import { - INGEST_DATABASE, - METRIC_STREAM_TABLE, -} from "../../metric-stream/clickhouse-table.ts"; import { buildPostgresFitnessActivityRawTableStatement } from "../clickhouse-raw-tables.ts"; import { buildActivityReadModelRefreshStatements } from "../clickhouse-read-models.ts"; import { @@ -194,10 +191,7 @@ export async function rebuildMetricStreamLocationPoint( client, "DROP TABLE IF EXISTS analytics.metric_stream_backfill_chunks", ); - await runClickHouseMigrationStatement( - client, - `DROP TABLE IF EXISTS ${METRIC_STREAM_TABLE}`, - ); + await runClickHouseMigrationStatement(client, `DROP TABLE IF EXISTS ${METRIC_STREAM_TABLE}`); } for (const statement of buildClickHouseBootstrapStatements(postgresConnectionString)) { @@ -315,8 +309,7 @@ async function shouldReplaceMetricStreamTable(client: ClickHouseCommandClient): throw new Error("ClickHouse migrations require a query-capable client"); } const result = await client.query({ - query: - `SELECT engine FROM system.tables WHERE database = '${INGEST_DATABASE}' AND name = 'metric_stream'`, + query: `SELECT engine FROM system.tables WHERE database = '${INGEST_DATABASE}' AND name = 'metric_stream'`, format: "JSONEachRow", }); const rows = await result.json(); diff --git a/src/db/clickhouse.ts b/src/db/clickhouse.ts index bb36c60885..b20602ee54 100644 --- a/src/db/clickhouse.ts +++ b/src/db/clickhouse.ts @@ -1,6 +1,6 @@ import { createClient } from "@clickhouse/client"; -import { buildClickHouseBootstrapStatementsForNativeMetricStream } from "./clickhouse-metric-stream-bootstrap.ts"; import { METRIC_STREAM_TABLE } from "../metric-stream/clickhouse-table.ts"; +import { buildClickHouseBootstrapStatementsForNativeMetricStream } from "./clickhouse-metric-stream-bootstrap.ts"; export interface ClickHouseCommandClient { command(options: { diff --git a/src/metric-stream/clickhouse-sink.test.ts b/src/metric-stream/clickhouse-sink.test.ts index 456222c2f2..4d314fecc4 100644 --- a/src/metric-stream/clickhouse-sink.test.ts +++ b/src/metric-stream/clickhouse-sink.test.ts @@ -1,11 +1,11 @@ import { afterEach, describe, expect, it, vi } from "vitest"; -import { METRIC_STREAM_TABLE } from "./clickhouse-table.ts"; import { applyMetricStreamEventsToClickHouse, insertMetricStreamEventsIntoClickHouse, mapMetricStreamEventToClickHouseRow, markMetricStreamScopeDeletedInClickHouse, } from "./clickhouse-sink.ts"; +import { METRIC_STREAM_TABLE } from "./clickhouse-table.ts"; import { createMetricStreamDeletedEvent, type MetricStreamEventV1 } from "./events.ts"; import type { RunMetricStreamEventConsumerOptions } from "./redpanda-consumer.ts"; @@ -32,14 +32,12 @@ const imuEvent = { channel: "imu", } satisfies MetricStreamEventV1; -function firstCommandQuery( - command: ReturnType, -): string { - const call = command.mock.calls[0]?.[0] as { query: string } | undefined; - if (!call) { +function firstCommandQuery(command: ReturnType): string { + const call = command.mock.calls[0]?.[0]; + if (!call || typeof call !== "object" || !("query" in call) || typeof call.query !== "string") { throw new Error("expected command call"); } - return String(call.query); + return call.query; } afterEach(() => { @@ -238,9 +236,7 @@ describe("applyMetricStreamEventsToClickHouse", () => { expect(query).toContain("external_id = {external_id:String}"); expect(query).toContain("channel = {channel:String}"); expect(query).toContain("toString(activity_id) = {activity_id:String}"); - expect(query).toContain( - "recorded_at >= parseDateTime64BestEffort({recorded_at_start:String})", - ); + expect(query).toContain("recorded_at >= parseDateTime64BestEffort({recorded_at_start:String})"); expect(query).toContain("recorded_at < parseDateTime64BestEffort({recorded_at_end:String})"); }); From c1044a150ed6ac78a6c04ed1287445b0182f23c2 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 17:38:08 -0700 Subject: [PATCH 05/11] Fix CI failures from metric stream migration and Stryker symlinks. Align integration test seeds and dbt microbatch event_time with ingest.metric_stream column names, and exclude agent skill symlinks from Stryker sandboxes. Co-authored-by: Cursor --- .../read_model_microbatch.sql.test.ts | 2 +- .../models/staging/sensor_scalar_sample.sql | 2 +- .../heart-rate-repository.integration.test.ts | 46 +++++++++---------- ...ider-detail-repository.integration.test.ts | 34 +++++++------- stryker.ci.config.json | 6 ++- stryker.config.json | 6 ++- 6 files changed, 52 insertions(+), 44 deletions(-) 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 80bd109892..a0de54d0a6 100644 --- a/analytics/models/read_models/read_model_microbatch.sql.test.ts +++ b/analytics/models/read_models/read_model_microbatch.sql.test.ts @@ -261,7 +261,7 @@ describe("production analytics read-model build", () => { expect(sourcesYaml).toContain("identifier: metric_stream"); expect(sourcesYaml).toContain("name: ingest"); expect(sourcesYaml).toContain("event_time: ingested_at"); - expect(sensorScalarSampleSql).toContain("event_time='ingested_at'"); + expect(sensorScalarSampleSql).toContain("event_time='_peerdb_synced_at'"); expect(sensorScalarSampleSql).toContain("source('ingest', 'metric_stream_freshness')"); expect(dedupedSensorSql).toContain("event_time='refreshed_at'"); expect(dedupedSensorSql).toContain("max(samples._peerdb_synced_at) AS source_refreshed_at"); diff --git a/analytics/models/staging/sensor_scalar_sample.sql b/analytics/models/staging/sensor_scalar_sample.sql index 853738eec9..5aed5c4e9c 100644 --- a/analytics/models/staging/sensor_scalar_sample.sql +++ b/analytics/models/staging/sensor_scalar_sample.sql @@ -2,7 +2,7 @@ materialized='incremental', incremental_strategy='microbatch', unique_key='id', - event_time='ingested_at', + event_time='_peerdb_synced_at', begin='2026-01-01', batch_size='day', lookback=3, diff --git a/packages/server/src/repositories/heart-rate-repository.integration.test.ts b/packages/server/src/repositories/heart-rate-repository.integration.test.ts index 672ab567d4..07a1f6a3fa 100644 --- a/packages/server/src/repositories/heart-rate-repository.integration.test.ts +++ b/packages/server/src/repositories/heart-rate-repository.integration.test.ts @@ -12,8 +12,8 @@ interface SeedRow { provider_id: string; channel: string; scalar: number; - _peerdb_is_deleted: 0 | 1; - _peerdb_version: number; + is_deleted: 0 | 1; + version: number; } const client = createClickHouseClientFromEnv(); @@ -38,9 +38,9 @@ async function seed(rows: SeedRow[]): Promise { 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, + is_deleted: row.is_deleted, + ingested_at: "2026-04-12 00:00:00.000", + version: row.version, })), }); } @@ -64,8 +64,8 @@ describe("HeartRateRepository (integration)", () => { provider_id: "whoop_ble", channel: "heart_rate", scalar: 72, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, { id: randomUUID(), @@ -73,8 +73,8 @@ describe("HeartRateRepository (integration)", () => { provider_id: "whoop_ble", channel: "heart_rate", scalar: 74, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, // apple_health: same minute as whoop — must stay a separate source (no priority collapse) { @@ -83,8 +83,8 @@ describe("HeartRateRepository (integration)", () => { provider_id: "apple_health", channel: "heart_rate", scalar: 70, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, // version dedup: same id/time, v1 supersedes v0 — FINAL must keep 99, not 50 { @@ -93,8 +93,8 @@ describe("HeartRateRepository (integration)", () => { provider_id: "whoop_ble", channel: "heart_rate", scalar: 50, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, { id: supersededId, @@ -102,8 +102,8 @@ describe("HeartRateRepository (integration)", () => { provider_id: "whoop_ble", channel: "heart_rate", scalar: 99, - _peerdb_is_deleted: 0, - _peerdb_version: 1, + is_deleted: 0, + version: 1, }, // excluded rows { @@ -112,8 +112,8 @@ describe("HeartRateRepository (integration)", () => { provider_id: "whoop_ble", channel: "heart_rate", scalar: 88, - _peerdb_is_deleted: 1, - _peerdb_version: 1, + is_deleted: 1, + version: 1, }, // deleted { id: randomUUID(), @@ -121,8 +121,8 @@ describe("HeartRateRepository (integration)", () => { provider_id: "whoop_ble", channel: "heart_rate", scalar: 60, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, // other day { id: randomUUID(), @@ -130,8 +130,8 @@ describe("HeartRateRepository (integration)", () => { provider_id: "whoop_ble", channel: "heart_rate", scalar: 0, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, // zero { id: randomUUID(), @@ -139,8 +139,8 @@ describe("HeartRateRepository (integration)", () => { provider_id: "whoop_ble", channel: "power", scalar: 200, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, // other channel ]); }, 120_000); diff --git a/packages/server/src/repositories/provider-detail-repository.integration.test.ts b/packages/server/src/repositories/provider-detail-repository.integration.test.ts index 0b1469e935..ab2d7fac96 100644 --- a/packages/server/src/repositories/provider-detail-repository.integration.test.ts +++ b/packages/server/src/repositories/provider-detail-repository.integration.test.ts @@ -22,8 +22,8 @@ interface SeedRow { provider_id: string; channel: string; scalar: number; - _peerdb_is_deleted: 0 | 1; - _peerdb_version: number; + is_deleted: 0 | 1; + version: number; } const client = createClickHouseClientFromEnv(); @@ -49,9 +49,9 @@ async function seed(rows: SeedRow[]): Promise { 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, + is_deleted: row.is_deleted, + ingested_at: "2026-04-12 00:00:00.000", + version: row.version, })), }); } @@ -82,8 +82,8 @@ describe("ProviderDetailRepository metric stream (integration)", () => { provider_id: "withings", channel: "heart_rate", scalar: 60, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, { id: randomUUID(), @@ -91,8 +91,8 @@ describe("ProviderDetailRepository metric stream (integration)", () => { provider_id: "withings", channel: "body_weight", scalar: 80, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, // version dedup: v1 (90) supersedes v0 (50) { @@ -101,8 +101,8 @@ describe("ProviderDetailRepository metric stream (integration)", () => { provider_id: "withings", channel: "body_weight", scalar: 50, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, { id: supersededId, @@ -110,8 +110,8 @@ describe("ProviderDetailRepository metric stream (integration)", () => { provider_id: "withings", channel: "body_weight", scalar: 90, - _peerdb_is_deleted: 0, - _peerdb_version: 1, + is_deleted: 0, + version: 1, }, // excluded: deleted + other provider { @@ -120,8 +120,8 @@ describe("ProviderDetailRepository metric stream (integration)", () => { provider_id: "withings", channel: "heart_rate", scalar: 70, - _peerdb_is_deleted: 1, - _peerdb_version: 1, + is_deleted: 1, + version: 1, }, { id: randomUUID(), @@ -129,8 +129,8 @@ describe("ProviderDetailRepository metric stream (integration)", () => { provider_id: "fitbit", channel: "heart_rate", scalar: 65, - _peerdb_is_deleted: 0, - _peerdb_version: 0, + is_deleted: 0, + version: 0, }, ]); }, 120_000); diff --git a/stryker.ci.config.json b/stryker.ci.config.json index 87fa00affa..2eeada5cd2 100644 --- a/stryker.ci.config.json +++ b/stryker.ci.config.json @@ -45,14 +45,18 @@ ], "ignorePatterns": [ "dist/**", + ".adal/**", + ".aider-desk/**", ".antigravitycli/**", + ".augment/**", ".codegraph/**", ".codex/**", ".claude/**", ".context/**", ".gemini/**", ".kilocode/**", - ".agents/**" + ".agents/**", + "**/skills/stripe-projects" ], "mutator": { "excludedMutations": ["StringLiteral"] diff --git a/stryker.config.json b/stryker.config.json index 7a5084fde5..323a2d630a 100644 --- a/stryker.config.json +++ b/stryker.config.json @@ -46,14 +46,18 @@ ], "ignorePatterns": [ "dist/**", + ".adal/**", + ".aider-desk/**", ".antigravitycli/**", + ".augment/**", ".codegraph/**", ".codex/**", ".claude/**", ".context/**", ".gemini/**", ".kilocode/**", - ".agents/**" + ".agents/**", + "**/skills/stripe-projects" ], "mutator": { "excludedMutations": ["StringLiteral"] From d8b805f051ffeb5e5c295a0fc7ff8bac86d56d78 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 17:50:34 -0700 Subject: [PATCH 06/11] Add tests for metric stream ingest migration to satisfy Stryker CI. Co-authored-by: Cursor --- .../0034_move_metric_stream_to_ingest.test.ts | 140 ++++++++++++++++++ 1 file changed, 140 insertions(+) create mode 100644 src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts diff --git a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts new file mode 100644 index 0000000000..f67f6d7c04 --- /dev/null +++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts @@ -0,0 +1,140 @@ +import { describe, expect, it, vi } from "vitest"; +import type { ClickHouseCommandClient } from "../clickhouse.ts"; +import { + INGEST_DATABASE, + LEGACY_METRIC_STREAM_TABLE, + METRIC_STREAM_TABLE, +} from "../../metric-stream/clickhouse-table.ts"; +import * as moveMetricStreamToIngestMigration from "./0034_move_metric_stream_to_ingest.ts"; + +function isLegacyMetricStreamTableCountQuery(queryText: string): boolean { + return ( + queryText.includes("system.tables") && + queryText.includes("postgres_fitness") && + queryText.includes("name = 'metric_stream'") && + queryText.includes("AS count") + ); +} + +function mockLegacyTableCountClient(countRows: unknown): ClickHouseCommandClient { + return { + query: vi.fn().mockImplementation(({ query: queryText }: { query: string }) => ({ + json: vi + .fn() + .mockResolvedValue( + isLegacyMetricStreamTableCountQuery(queryText) ? countRows : [{ migration_count: 0 }], + ), + })), + command: vi.fn().mockResolvedValue(undefined), + }; +} + +describe("0034_move_metric_stream_to_ingest", () => { + it("creates the ingest database and metric stream table", () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + + expect(migration.id).toBe("0034_move_metric_stream_to_ingest"); + expect(migration.statements).toEqual([ + `CREATE DATABASE IF NOT EXISTS ${INGEST_DATABASE}`, + expect.stringContaining(`CREATE TABLE IF NOT EXISTS ${METRIC_STREAM_TABLE}`), + ]); + }); + + it("rejects a client without query method", async () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + const client: ClickHouseCommandClient = { command: vi.fn() }; + + await expect(migration.run?.(client, "postgres://health:fixture@db:5432/health")).rejects.toThrow( + "ClickHouse migrations require a query-capable client", + ); + }); + + it("skips copy and drop when the legacy mirror table is absent", async () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + const client = mockLegacyTableCountClient([{ count: "0" }]); + + await migration.run?.(client, "postgres://health:fixture@db:5432/health"); + + expect(client.command).not.toHaveBeenCalled(); + }); + + it("skips copy and drop when the legacy table count query returns no rows", async () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + const client = mockLegacyTableCountClient([]); + + await migration.run?.(client, "postgres://health:fixture@db:5432/health"); + + expect(client.command).not.toHaveBeenCalled(); + }); + + it("skips copy and drop when the legacy table count response is not an array", async () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + const client = mockLegacyTableCountClient({ 0: { count: "1" } }); + + await migration.run?.(client, "postgres://health:fixture@db:5432/health"); + + expect(client.command).not.toHaveBeenCalled(); + }); + + it("skips copy and drop when the legacy table count row is not a string count", async () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + const client = mockLegacyTableCountClient([{ count: 1 }]); + + await migration.run?.(client, "postgres://health:fixture@db:5432/health"); + + expect(client.command).not.toHaveBeenCalled(); + }); + + it("skips copy and drop when the legacy table count row is missing the count field", async () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + const client = mockLegacyTableCountClient([{ table_count: "1" }]); + + await migration.run?.(client, "postgres://health:fixture@db:5432/health"); + + expect(client.command).not.toHaveBeenCalled(); + }); + + it("skips copy and drop when the legacy table count row is null", async () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + const client = mockLegacyTableCountClient([null]); + + await migration.run?.(client, "postgres://health:fixture@db:5432/health"); + + expect(client.command).not.toHaveBeenCalled(); + }); + + it("copies legacy metric stream rows into ingest and drops the legacy mirror", async () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + const client = mockLegacyTableCountClient([{ count: "1" }]); + + await migration.run?.(client, "postgres://health:fixture@db:5432/health"); + + expect(client.command).toHaveBeenCalledTimes(2); + expect(client.command).toHaveBeenNthCalledWith(1, { + query: expect.stringContaining(`INSERT INTO ${METRIC_STREAM_TABLE}`), + }); + expect(client.command).toHaveBeenNthCalledWith(1, { + query: expect.stringContaining(`FROM ${LEGACY_METRIC_STREAM_TABLE}`), + }); + expect(client.command).toHaveBeenNthCalledWith(1, { + query: expect.stringContaining("_peerdb_synced_at AS ingested_at"), + }); + expect(client.command).toHaveBeenNthCalledWith(2, { + query: `DROP TABLE IF EXISTS ${LEGACY_METRIC_STREAM_TABLE}`, + }); + }); + + it("copies legacy metric stream rows when the legacy table count is greater than zero", async () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + const client = mockLegacyTableCountClient([{ count: "2" }]); + + await migration.run?.(client, "postgres://health:fixture@db:5432/health"); + + expect(client.command).toHaveBeenCalledWith({ + query: expect.stringContaining(`INSERT INTO ${METRIC_STREAM_TABLE}`), + }); + expect(client.command).toHaveBeenCalledWith({ + query: `DROP TABLE IF EXISTS ${LEGACY_METRIC_STREAM_TABLE}`, + }); + }); +}); From c362ba81404e94c66ca0c6da92a3613e5b8ecc57 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 17:57:04 -0700 Subject: [PATCH 07/11] Fix Biome lint issues in metric stream ingest migration tests. Co-authored-by: Cursor --- .../0034_move_metric_stream_to_ingest.test.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts index f67f6d7c04..474b4af8f1 100644 --- a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts +++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts @@ -1,10 +1,10 @@ import { describe, expect, it, vi } from "vitest"; -import type { ClickHouseCommandClient } from "../clickhouse.ts"; import { INGEST_DATABASE, LEGACY_METRIC_STREAM_TABLE, METRIC_STREAM_TABLE, } from "../../metric-stream/clickhouse-table.ts"; +import type { ClickHouseCommandClient } from "../clickhouse.ts"; import * as moveMetricStreamToIngestMigration from "./0034_move_metric_stream_to_ingest.ts"; function isLegacyMetricStreamTableCountQuery(queryText: string): boolean { @@ -44,9 +44,9 @@ describe("0034_move_metric_stream_to_ingest", () => { const migration = moveMetricStreamToIngestMigration.createMigration(); const client: ClickHouseCommandClient = { command: vi.fn() }; - await expect(migration.run?.(client, "postgres://health:fixture@db:5432/health")).rejects.toThrow( - "ClickHouse migrations require a query-capable client", - ); + await expect( + migration.run?.(client, "postgres://health:fixture@db:5432/health"), + ).rejects.toThrow("ClickHouse migrations require a query-capable client"); }); it("skips copy and drop when the legacy mirror table is absent", async () => { From 5ba865150d28697b731be64a3fb76c9ab7552528 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 21:30:32 -0700 Subject: [PATCH 08/11] Address CodeRabbit review feedback for metric stream ingest migration. Co-authored-by: Cursor --- .../heart-rate-repository.integration.test.ts | 32 +++--------- .../src/repositories/heart-rate-repository.ts | 4 +- .../metric-stream-integration-test-helpers.ts | 39 ++++++++++++++ ...ider-detail-repository.integration.test.ts | 32 +++--------- .../provider-detail-repository.test.ts | 2 - .../provider-detail-repository.ts | 4 +- scripts/seed-review-clickhouse.test.ts | 4 ++ scripts/seed-review-clickhouse.ts | 30 +++++++++-- .../0034_move_metric_stream_to_ingest.test.ts | 51 ++++--------------- .../0034_move_metric_stream_to_ingest.ts | 26 ++++------ src/providers/garmin/sync-step-plan.test.ts | 38 +++++++++----- src/providers/garmin/sync-step-plan.ts | 13 ++--- 12 files changed, 139 insertions(+), 136 deletions(-) create mode 100644 packages/server/src/repositories/metric-stream-integration-test-helpers.ts diff --git a/packages/server/src/repositories/heart-rate-repository.integration.test.ts b/packages/server/src/repositories/heart-rate-repository.integration.test.ts index 07a1f6a3fa..712cda32f2 100644 --- a/packages/server/src/repositories/heart-rate-repository.integration.test.ts +++ b/packages/server/src/repositories/heart-rate-repository.integration.test.ts @@ -3,19 +3,13 @@ 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"; +import { + type MetricStreamSeedRow, + seedMetricStreamRows, +} from "./metric-stream-integration-test-helpers.ts"; const testUserId = "00000000-0000-0000-0000-0000000000a1"; -interface SeedRow { - id: string; - recorded_at: string; - provider_id: string; - channel: string; - scalar: number; - is_deleted: 0 | 1; - version: number; -} - const client = createClickHouseClientFromEnv(); const reader: MetricStreamClickHouseReader = { @@ -27,22 +21,8 @@ const reader: MetricStreamClickHouseReader = { }, }; -async function seed(rows: SeedRow[]): Promise { - await client.insert?.({ - table: "ingest.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, - is_deleted: row.is_deleted, - ingested_at: "2026-04-12 00:00:00.000", - version: row.version, - })), - }); +async function seed(rows: MetricStreamSeedRow[]): Promise { + await seedMetricStreamRows(client, testUserId, rows); } describe("HeartRateRepository (integration)", () => { diff --git a/packages/server/src/repositories/heart-rate-repository.ts b/packages/server/src/repositories/heart-rate-repository.ts index 401768638b..142b879034 100644 --- a/packages/server/src/repositories/heart-rate-repository.ts +++ b/packages/server/src/repositories/heart-rate-repository.ts @@ -43,7 +43,7 @@ export class HeartRateRepository { * * 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 + * version-deduplicated (`FINAL` + `is_deleted = 0`) but NOT collapsed * by provider priority, so every source is returned for overlay/comparison. */ async dailyBySource(date: string): Promise { @@ -62,7 +62,7 @@ export class HeartRateRepository { toStartOfMinute(toDateTime(recorded_at)) AS minute_bucket, scalar FROM ingest.metric_stream FINAL - WHERE user_id = {userId:String} + WHERE user_id = {userId:UUID} AND channel = 'heart_rate' AND is_deleted = 0 AND scalar > 0 diff --git a/packages/server/src/repositories/metric-stream-integration-test-helpers.ts b/packages/server/src/repositories/metric-stream-integration-test-helpers.ts new file mode 100644 index 0000000000..d216ffa484 --- /dev/null +++ b/packages/server/src/repositories/metric-stream-integration-test-helpers.ts @@ -0,0 +1,39 @@ +export interface MetricStreamSeedRow { + id: string; + recorded_at: string; + provider_id: string; + channel: string; + scalar: number; + is_deleted: 0 | 1; + version: number; +} + +interface MetricStreamSeedClient { + insert?: (args: { + table: string; + format: "JSONEachRow"; + values: Record[]; + }) => Promise; +} + +export async function seedMetricStreamRows( + client: MetricStreamSeedClient, + userId: string, + rows: MetricStreamSeedRow[], +): Promise { + await client.insert?.({ + table: "ingest.metric_stream", + format: "JSONEachRow", + values: rows.map((row) => ({ + id: row.id, + user_id: userId, + recorded_at: row.recorded_at, + provider_id: row.provider_id, + channel: row.channel, + scalar: row.scalar, + is_deleted: row.is_deleted, + ingested_at: "2026-04-12 00:00:00.000", + version: row.version, + })), + }); +} diff --git a/packages/server/src/repositories/provider-detail-repository.integration.test.ts b/packages/server/src/repositories/provider-detail-repository.integration.test.ts index ab2d7fac96..0851d84549 100644 --- a/packages/server/src/repositories/provider-detail-repository.integration.test.ts +++ b/packages/server/src/repositories/provider-detail-repository.integration.test.ts @@ -5,6 +5,10 @@ import { z } from "zod"; import { createClickHouseClientFromEnv } from "../../../../src/db/clickhouse.ts"; import { buildClickHouseBootstrapStatementsForNativeMetricStream } from "../../../../src/db/clickhouse-metric-stream-bootstrap.ts"; import type { BodyClickHouseStore } from "./body-clickhouse.ts"; +import { + type MetricStreamSeedRow, + seedMetricStreamRows, +} from "./metric-stream-integration-test-helpers.ts"; import { ProviderDetailRepository } from "./provider-detail-repository.ts"; // metricStream reads only touch ClickHouse; the repository still requires a @@ -16,16 +20,6 @@ const noopDb: Pick = { const testUserId = "00000000-0000-0000-0000-0000000000b2"; -interface SeedRow { - id: string; - recorded_at: string; - provider_id: string; - channel: string; - scalar: number; - is_deleted: 0 | 1; - version: number; -} - const client = createClickHouseClientFromEnv(); // The repository only needs a ClickHouse query runner (BodyClickHouseStore shape). @@ -38,22 +32,8 @@ const clickHouse: BodyClickHouseStore = { }, }; -async function seed(rows: SeedRow[]): Promise { - await client.insert?.({ - table: "ingest.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, - is_deleted: row.is_deleted, - ingested_at: "2026-04-12 00:00:00.000", - version: row.version, - })), - }); +async function seed(rows: MetricStreamSeedRow[]): Promise { + await seedMetricStreamRows(client, testUserId, rows); } const rowSchema = z.object({ diff --git a/packages/server/src/repositories/provider-detail-repository.test.ts b/packages/server/src/repositories/provider-detail-repository.test.ts index 78026df852..d20c7f4090 100644 --- a/packages/server/src/repositories/provider-detail-repository.test.ts +++ b/packages/server/src/repositories/provider-detail-repository.test.ts @@ -203,7 +203,6 @@ describe("ProviderDetailRepository", () => { expect(query.mock.calls[0]?.[2]).toStrictEqual({ userId: "user-1", providerId: "withings", - recordId: "", limit: 10, offset: 5, }); @@ -233,7 +232,6 @@ describe("ProviderDetailRepository", () => { expect(query.mock.calls[0]?.[2]).toStrictEqual({ userId: "user-1", providerId: "whoop", - recordId: "", limit: 10, offset: 5, }); diff --git a/packages/server/src/repositories/provider-detail-repository.ts b/packages/server/src/repositories/provider-detail-repository.ts index 277d07a5f5..cd5e2fb60f 100644 --- a/packages/server/src/repositories/provider-detail-repository.ts +++ b/packages/server/src/repositories/provider-detail-repository.ts @@ -325,7 +325,7 @@ export class ProviderDetailRepository { { userId: this.#userId, providerId, - recordId: recordId ?? "", + ...(recordId ? { recordId } : {}), limit, offset, }, @@ -375,7 +375,7 @@ export class ProviderDetailRepository { { userId: this.#userId, providerId, - recordId: recordId ?? "", + ...(recordId ? { recordId } : {}), limit, offset, }, diff --git a/scripts/seed-review-clickhouse.test.ts b/scripts/seed-review-clickhouse.test.ts index 2a4aa7855e..d49edd1562 100644 --- a/scripts/seed-review-clickhouse.test.ts +++ b/scripts/seed-review-clickhouse.test.ts @@ -23,6 +23,10 @@ describe("seed-review-clickhouse", () => { expect(statements).toContain("TRUNCATE TABLE IF EXISTS ingest.metric_stream"); expect(statements.join("\n")).toContain("INSERT INTO postgres_fitness.activity"); expect(statements.join("\n")).toContain("INSERT INTO ingest.metric_stream"); + expect(statements.join("\n")).toContain("ingested_at, is_deleted, version"); + expect(statements.join("\n")).not.toContain( + "INSERT INTO ingest.metric_stream (recorded_at, user_id, provider_id, external_id, device_id, source_type, channel, activity_id, scalar, point, id, _peerdb_synced_at", + ); expect(statements.join("\n")).toContain( "WHERE user_id = '00000000-0000-0000-0000-000000000001'", ); diff --git a/scripts/seed-review-clickhouse.ts b/scripts/seed-review-clickhouse.ts index d22b9e1272..a2b4ec7db7 100644 --- a/scripts/seed-review-clickhouse.ts +++ b/scripts/seed-review-clickhouse.ts @@ -10,9 +10,29 @@ interface ReviewRawTableCopy { sourceTable: string; columns: readonly string[]; selectExpressions: readonly string[]; + metadataColumns?: readonly string[]; + metadataExpressions?: readonly string[]; whereClause?: string; } +const peerDbMetadataColumns = [ + "_peerdb_synced_at", + "_peerdb_is_deleted", + "_peerdb_version", +] as const; +const peerDbMetadataExpressions = [ + "now64(9)", + "0", + "toInt64(toUnixTimestamp64Micro(now64(6)))", +] as const; + +const ingestMetricStreamMetadataColumns = ["ingested_at", "is_deleted", "version"] as const; +const ingestMetricStreamMetadataExpressions = [ + "now64(9)", + "0", + "toInt64(toUnixTimestamp64Micro(now64(6)))", +] as const; + const seedProviderIds = ["whoop", "apple_health", "strava", "bodyspec", "manual_review"]; const seedProviderList = seedProviderIds.map(clickHouseStringLiteral).join(", "); @@ -180,6 +200,8 @@ const reviewRawTableCopies: readonly ReviewRawTableCopy[] = [ "id", ], whereClause: `user_id = ${clickHouseStringLiteral(USER_ID)}`, + metadataColumns: ingestMetricStreamMetadataColumns, + metadataExpressions: ingestMetricStreamMetadataExpressions, }, ]; @@ -229,10 +251,10 @@ function buildCopyStatement( tableCopy: ReviewRawTableCopy, ): string { const source = buildReviewPostgresTableFunction(postgresConnectionString, tableCopy.sourceTable); - const peerDbColumns = ["_peerdb_synced_at", "_peerdb_is_deleted", "_peerdb_version"]; - const peerDbExpressions = ["now64(9)", "0", "toInt64(toUnixTimestamp64Micro(now64(6)))"]; - const allColumns = [...tableCopy.columns, ...peerDbColumns]; - const allExpressions = [...tableCopy.selectExpressions, ...peerDbExpressions]; + const metadataColumns = tableCopy.metadataColumns ?? peerDbMetadataColumns; + const metadataExpressions = tableCopy.metadataExpressions ?? peerDbMetadataExpressions; + const allColumns = [...tableCopy.columns, ...metadataColumns]; + const allExpressions = [...tableCopy.selectExpressions, ...metadataExpressions]; const whereClause = tableCopy.whereClause ? `\nWHERE ${tableCopy.whereClause}` : ""; return `INSERT INTO ${tableCopy.targetTable} (${allColumns.join(", ")}) diff --git a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts index 474b4af8f1..86867b1a92 100644 --- a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts +++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts @@ -58,48 +58,19 @@ describe("0034_move_metric_stream_to_ingest", () => { expect(client.command).not.toHaveBeenCalled(); }); - it("skips copy and drop when the legacy table count query returns no rows", async () => { + it.each([ + { label: "no rows", countRows: [] }, + { label: "a non-array response", countRows: { 0: { count: "1" } } }, + { label: "a non-string count", countRows: [{ count: 1 }] }, + { label: "a missing count field", countRows: [{ table_count: "1" }] }, + { label: "a null row", countRows: [null] }, + ])("rejects malformed legacy table count payloads ($label)", async ({ countRows }) => { const migration = moveMetricStreamToIngestMigration.createMigration(); - const client = mockLegacyTableCountClient([]); - - await migration.run?.(client, "postgres://health:fixture@db:5432/health"); - - expect(client.command).not.toHaveBeenCalled(); - }); - - it("skips copy and drop when the legacy table count response is not an array", async () => { - const migration = moveMetricStreamToIngestMigration.createMigration(); - const client = mockLegacyTableCountClient({ 0: { count: "1" } }); - - await migration.run?.(client, "postgres://health:fixture@db:5432/health"); - - expect(client.command).not.toHaveBeenCalled(); - }); - - it("skips copy and drop when the legacy table count row is not a string count", async () => { - const migration = moveMetricStreamToIngestMigration.createMigration(); - const client = mockLegacyTableCountClient([{ count: 1 }]); - - await migration.run?.(client, "postgres://health:fixture@db:5432/health"); - - expect(client.command).not.toHaveBeenCalled(); - }); - - it("skips copy and drop when the legacy table count row is missing the count field", async () => { - const migration = moveMetricStreamToIngestMigration.createMigration(); - const client = mockLegacyTableCountClient([{ table_count: "1" }]); - - await migration.run?.(client, "postgres://health:fixture@db:5432/health"); - - expect(client.command).not.toHaveBeenCalled(); - }); - - it("skips copy and drop when the legacy table count row is null", async () => { - const migration = moveMetricStreamToIngestMigration.createMigration(); - const client = mockLegacyTableCountClient([null]); - - await migration.run?.(client, "postgres://health:fixture@db:5432/health"); + const client = mockLegacyTableCountClient(countRows); + await expect( + migration.run?.(client, "postgres://health:fixture@db:5432/health"), + ).rejects.toThrow(); expect(client.command).not.toHaveBeenCalled(); }); diff --git a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts index 2d81e8fe30..f0613ddf64 100644 --- a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts +++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts @@ -1,3 +1,4 @@ +import { z } from "zod"; import { buildIngestMetricStreamCreateTableSql, INGEST_DATABASE, @@ -7,24 +8,17 @@ import { import type { ClickHouseCommandClient } from "../clickhouse.ts"; import type { ClickHouseMigration } from "./types.ts"; -function isLegacyTableCountRow(value: unknown): value is { count: string } { - return ( - typeof value === "object" && - value !== null && - "count" in value && - typeof value.count === "string" - ); -} +const legacyTableCountRowsSchema = z + .array( + z.object({ + count: z.string().regex(/^\d+$/), + }), + ) + .min(1, "Expected at least one row from system.tables count query"); function parseLegacyTableCount(rows: unknown): number { - if (!Array.isArray(rows) || rows.length === 0) { - return 0; - } - const first = rows[0]; - if (!isLegacyTableCountRow(first)) { - return 0; - } - return Number(first.count); + const parsedRows = legacyTableCountRowsSchema.parse(rows); + return Number(parsedRows[0].count); } async function legacyMetricStreamTableExists(client: ClickHouseCommandClient): Promise { diff --git a/src/providers/garmin/sync-step-plan.test.ts b/src/providers/garmin/sync-step-plan.test.ts index 26265d8826..1ff4ce38c2 100644 --- a/src/providers/garmin/sync-step-plan.test.ts +++ b/src/providers/garmin/sync-step-plan.test.ts @@ -23,6 +23,10 @@ const clickHouseMocks = vi.hoisted(() => { }; }); +const sentryMocks = vi.hoisted(() => ({ + captureException: vi.fn(), +})); + vi.mock("../../db/token-user-context.ts", () => ({ getTokenUserId: tokenUserContextMocks.getTokenUserId, })); @@ -31,6 +35,10 @@ vi.mock("../../db/clickhouse.ts", () => ({ createClickHouseClientFromEnv: clickHouseMocks.createClickHouseClientFromEnv, })); +vi.mock("@sentry/node", () => ({ + captureException: sentryMocks.captureException, +})); + function makeDb(selectedRows: unknown[] = []) { const chain = { values: vi.fn(), @@ -82,6 +90,7 @@ beforeEach(() => { clickHouseMocks.query.mockReset(); clickHouseMocks.query.mockResolvedValue({ json: async () => [] }); clickHouseMocks.close.mockClear(); + sentryMocks.captureException.mockClear(); delete process.env.CLICKHOUSE_URL; }); @@ -245,9 +254,14 @@ describe("Garmin sync step planning", () => { endedAt: new Date("2026-03-02T07:00:00.000Z"), }, ]); - clickHouseMocks.query.mockResolvedValue({ - json: async () => [{ date: "2026-03-01" }, { date: "2026-03-02" }], - }); + clickHouseMocks.query.mockImplementation( + async (args: { query_params?: { channel?: string } }) => ({ + json: async () => + args.query_params?.channel === "stress" + ? [{ date: "2026-03-01" }, { date: "2026-03-02" }] + : [], + }), + ); const context = makeContext({ db: db.db, userId: "options-user" }); await expect(planGarminSyncSteps(context)).resolves.toEqual([ @@ -269,20 +283,20 @@ describe("Garmin sync step planning", () => { }, }), ); - expect(clickHouseMocks.close).toHaveBeenCalledTimes(1); + expect(clickHouseMocks.close).toHaveBeenCalledTimes(2); }); - it("closes ClickHouse clients when metric-stream queries fail", async () => { + it("closes ClickHouse clients and surfaces metric-stream query failures", async () => { process.env.CLICKHOUSE_URL = "http://default:health@127.0.0.1:8123"; - clickHouseMocks.query.mockRejectedValue(new Error("connection refused")); + const queryError = new Error("connection refused"); + clickHouseMocks.query.mockRejectedValue(queryError); const context = makeContext(); - await expect(planGarminSyncSteps(context)).resolves.toEqual( - expect.arrayContaining([ - { type: "stress", date: "2026-03-01" }, - { type: "heart_rate", date: "2026-03-01" }, - ]), - ); + await expect(planGarminSyncSteps(context)).rejects.toThrow("connection refused"); expect(clickHouseMocks.close).toHaveBeenCalled(); + expect(sentryMocks.captureException).toHaveBeenCalledWith(queryError, { + tags: { provider: "garmin", operation: "metric_stream_date_query" }, + extra: { channel: "stress", providerId: "garmin" }, + }); }); }); diff --git a/src/providers/garmin/sync-step-plan.ts b/src/providers/garmin/sync-step-plan.ts index 103689e991..1a98cc61d3 100644 --- a/src/providers/garmin/sync-step-plan.ts +++ b/src/providers/garmin/sync-step-plan.ts @@ -1,10 +1,11 @@ +import { captureException } from "@sentry/node"; import { and, eq, gte, isNotNull, lt, lte, or } from "drizzle-orm"; import { z } from "zod"; +import { createClickHouseClientFromEnv } from "../../db/clickhouse.ts"; import type { SyncDatabase } from "../../db/index.ts"; import { dailyMetrics, sleepSession } from "../../db/schema.ts"; import { HEART_RATE, STRESS } from "../../db/sensor-channels.ts"; import { getTokenUserId } from "../../db/token-user-context.ts"; -import { logger } from "../../logger.ts"; import type { GarminSyncStep } from "./sync-checkpoint.ts"; export interface GarminSyncPlanContext { @@ -138,7 +139,6 @@ async function listGarminDatesWithMetricStreamChannel( return new Set(); } - const { createClickHouseClientFromEnv } = await import("../../db/clickhouse.ts"); const client = createClickHouseClientFromEnv(); try { const result = await client.query({ @@ -164,10 +164,11 @@ async function listGarminDatesWithMetricStreamChannel( const rows = z.array(metricStreamDateRowSchema).parse(await result.json()); return new Set(rows.map((row) => row.date)); } catch (error) { - logger.warn( - `[garmin] ClickHouse metric-stream query failed for ${channel}: ${error instanceof Error ? error.message : String(error)}`, - ); - return new Set(); + captureException(error, { + tags: { provider: "garmin", operation: "metric_stream_date_query" }, + extra: { channel, providerId: context.providerId }, + }); + throw error; } finally { await client.close?.(); } From ebf3eedf217ad5d84b2684dbd583857a68647e1a Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 21:35:51 -0700 Subject: [PATCH 09/11] Fix TypeScript error in legacy table count Zod parsing. Co-authored-by: Cursor --- .../0034_move_metric_stream_to_ingest.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts index f0613ddf64..d9070b1f0a 100644 --- a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts +++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts @@ -14,11 +14,11 @@ const legacyTableCountRowsSchema = z count: z.string().regex(/^\d+$/), }), ) - .min(1, "Expected at least one row from system.tables count query"); + .nonempty("Expected at least one row from system.tables count query"); function parseLegacyTableCount(rows: unknown): number { - const parsedRows = legacyTableCountRowsSchema.parse(rows); - return Number(parsedRows[0].count); + const [firstRow] = legacyTableCountRowsSchema.parse(rows); + return Number(firstRow.count); } async function legacyMetricStreamTableExists(client: ClickHouseCommandClient): Promise { From 17ec99f56e80e1308d1c1af9844ca4180f6911a1 Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 21:46:54 -0700 Subject: [PATCH 10/11] Fix CI test failures for migration 0034 and Garmin sync planning. Align ClickHouse migration mocks with the new count-field Zod validation and mock ClickHouse in Garmin tests so CI does not attempt real connections. Co-authored-by: Cursor --- src/db/clickhouse-migrations.test.ts | 45 ++++++++++++-------- src/providers/garmin.test.ts | 19 +++++++++ src/providers/garmin/sync-checkpoint.test.ts | 24 ++++++++++- 3 files changed, 69 insertions(+), 19 deletions(-) diff --git a/src/db/clickhouse-migrations.test.ts b/src/db/clickhouse-migrations.test.ts index ec77e28660..617ad8a64a 100644 --- a/src/db/clickhouse-migrations.test.ts +++ b/src/db/clickhouse-migrations.test.ts @@ -30,6 +30,15 @@ import * as activityUserSoftDeleteMigration from "./clickhouse-migrations/0031_a import * as dedupedActivitiesAbsentSourceLinksMigration from "./clickhouse-migrations/0032_deduped_activities_absent_source_links.ts"; import { clickHouseMigrationFileNames } from "./clickhouse-migrations/registry.ts"; import { runClickHouseMigrationStatement } from "./clickhouse-migrations/statement-runner.ts"; + +function systemTablesQueryRows( + queryText: string, +): Array<{ count: string } | { table_count: number }> { + if (queryText.includes("AS count")) { + return [{ count: "1" }]; + } + return [{ table_count: 1 }]; +} import { buildClickHouseMigrationStatements, runClickHouseMigrations, @@ -398,7 +407,7 @@ SETTINGS allow_nullable_key = 1`); }; } if (queryText.includes("system.tables")) { - return { json: vi.fn().mockResolvedValue([{ table_count: 1 }]) }; + return { json: vi.fn().mockResolvedValue(systemTablesQueryRows(queryText)) }; } if (queryText.includes("min_recorded_at_ms")) { return { @@ -451,7 +460,7 @@ describe("runClickHouseMigrations", () => { return { json: vi.fn().mockResolvedValue([{ engine: "ReplacingMergeTree" }]) }; } if (queryText.includes("system.tables")) { - return { json: vi.fn().mockResolvedValue([{ table_count: 1 }]) }; + return { json: vi.fn().mockResolvedValue(systemTablesQueryRows(queryText)) }; } if (queryText.includes("system.databases")) { return { json: vi.fn().mockResolvedValue([{ engine: "Atomic" }]) }; @@ -539,7 +548,7 @@ describe("runClickHouseMigrations", () => { const command = vi.fn().mockResolvedValue(undefined); const query = vi.fn().mockImplementation(({ query: queryText }: { query: string }) => { if (queryText.includes("system.tables")) { - return { json: vi.fn().mockResolvedValue([{ table_count: 1 }]) }; + return { json: vi.fn().mockResolvedValue(systemTablesQueryRows(queryText)) }; } if (queryText.includes("0020_incremental_deduped_sensor")) { return { json: vi.fn().mockResolvedValue([{ migration_count: 0 }]) }; @@ -613,7 +622,7 @@ describe("runClickHouseMigrations", () => { queryText.includes("system.columns") ? [{ migration_count: 5 }] : queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0006_backfill_native_metric_stream") @@ -723,7 +732,7 @@ describe("runClickHouseMigrations", () => { queryText.includes("system.tables") && queryText.includes("engine") ? [{ engine: "ReplacingMergeTree" }] : queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0006_backfill_native_metric_stream") || @@ -780,7 +789,7 @@ describe("runClickHouseMigrations", () => { queryText.includes("system.columns") ? [{ migration_count: 5 }] : queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0012_repair_metric_stream_backfill") @@ -818,7 +827,7 @@ describe("runClickHouseMigrations", () => { queryText.includes("system.columns") ? [{ migration_count: 0 }] : queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0012_repair_metric_stream_backfill") @@ -876,7 +885,7 @@ describe("runClickHouseMigrations", () => { .fn() .mockResolvedValue( queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0013_metric_stream_location_point") @@ -942,7 +951,7 @@ describe("runClickHouseMigrations", () => { .fn() .mockResolvedValue( queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.columns") ? [{ migration_count: 5 }] : queryText.includes("system.databases") @@ -1014,7 +1023,7 @@ describe("runClickHouseMigrations", () => { .fn() .mockResolvedValue( queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0006_backfill_native_metric_stream") @@ -1066,7 +1075,7 @@ describe("runClickHouseMigrations", () => { .fn() .mockResolvedValue( queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0006_backfill_native_metric_stream") @@ -1104,7 +1113,7 @@ describe("runClickHouseMigrations", () => { .fn() .mockResolvedValue( queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0006_backfill_native_metric_stream") @@ -1154,7 +1163,7 @@ describe("runClickHouseMigrations", () => { return { json: vi.fn().mockResolvedValue([{ engine: "MergeTree" }]) }; } if (queryText.includes("system.tables")) { - return { json: vi.fn().mockResolvedValue([{ table_count: 1 }]) }; + return { json: vi.fn().mockResolvedValue(systemTablesQueryRows(queryText)) }; } if (queryText.includes("system.databases")) { return { json: vi.fn().mockResolvedValue([{ engine: "Atomic" }]) }; @@ -1208,7 +1217,7 @@ describe("runClickHouseMigrations", () => { .fn() .mockResolvedValue( queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "PostgreSQL" }] : queryText.includes("0006_backfill_native_metric_stream") @@ -1258,7 +1267,7 @@ describe("runClickHouseMigrations", () => { .fn() .mockResolvedValue( queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0006_backfill_native_metric_stream") @@ -1325,7 +1334,7 @@ describe("runClickHouseMigrations", () => { const query = vi.fn().mockImplementation(({ query: queryText }: { query: string }) => ({ json: vi.fn().mockResolvedValue( queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0006_backfill_native_metric_stream") @@ -1368,7 +1377,7 @@ describe("runClickHouseMigrations", () => { .fn() .mockResolvedValue( queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0006_backfill_native_metric_stream") @@ -1423,7 +1432,7 @@ describe("runClickHouseMigrations", () => { .fn() .mockResolvedValue( queryText.includes("system.tables") - ? [{ table_count: 1 }] + ? systemTablesQueryRows(queryText) : queryText.includes("system.databases") ? [{ engine: "Atomic" }] : queryText.includes("0006_backfill_native_metric_stream") diff --git a/src/providers/garmin.test.ts b/src/providers/garmin.test.ts index bd400071ad..266527c57a 100644 --- a/src/providers/garmin.test.ts +++ b/src/providers/garmin.test.ts @@ -83,6 +83,20 @@ const providerActivityAbsenceMocks = vi.hoisted(() => ({ upsertProviderActivity: vi.fn().mockResolvedValue({ id: "activity-id" }), })); +const clickHouseMocks = vi.hoisted(() => { + const query = vi.fn(); + const close = vi.fn(); + return { + query, + close, + createClickHouseClientFromEnv: vi.fn(() => ({ query, close })), + }; +}); + +vi.mock("../db/clickhouse.ts", () => ({ + createClickHouseClientFromEnv: clickHouseMocks.createClickHouseClientFromEnv, +})); + vi.mock("@sentry/node", () => ({ captureException: vi.fn(), })); @@ -554,6 +568,11 @@ describe("GarminProvider.sync()", () => { beforeEach(() => { vi.clearAllMocks(); publishedMetricStreamBatches.length = 0; + delete process.env.CLICKHOUSE_URL; + clickHouseMocks.createClickHouseClientFromEnv.mockClear(); + clickHouseMocks.query.mockReset(); + clickHouseMocks.query.mockResolvedValue({ json: async () => [] }); + clickHouseMocks.close.mockClear(); providerActivityAbsenceMocks.markProviderActivityAbsent.mockClear(); providerActivityAbsenceMocks.finishProviderActivityListSync.mockClear(); providerActivityAbsenceMocks.upsertProviderActivity.mockResolvedValue({ diff --git a/src/providers/garmin/sync-checkpoint.test.ts b/src/providers/garmin/sync-checkpoint.test.ts index f8e22bcd8f..a71ce2a920 100644 --- a/src/providers/garmin/sync-checkpoint.test.ts +++ b/src/providers/garmin/sync-checkpoint.test.ts @@ -1,5 +1,19 @@ -import { describe, expect, it, vi } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; import type { SyncDatabase } from "../../db/index.ts"; + +const clickHouseMocks = vi.hoisted(() => { + const query = vi.fn(); + const close = vi.fn(); + return { + query, + close, + createClickHouseClientFromEnv: vi.fn(() => ({ query, close })), + }; +}); + +vi.mock("../../db/clickhouse.ts", () => ({ + createClickHouseClientFromEnv: clickHouseMocks.createClickHouseClientFromEnv, +})); import { applyRateLimitToCheckpoint, createGarminSyncCheckpoint, @@ -9,6 +23,14 @@ import { } from "./sync-checkpoint.ts"; import { planGarminSyncSteps } from "./sync-step-plan.ts"; +beforeEach(() => { + clickHouseMocks.createClickHouseClientFromEnv.mockClear(); + clickHouseMocks.query.mockReset(); + clickHouseMocks.query.mockResolvedValue({ json: async () => [] }); + clickHouseMocks.close.mockClear(); + delete process.env.CLICKHOUSE_URL; +}); + function makeDb(): SyncDatabase { const chain = { from: vi.fn(), From 0b4e2d00dc0003366b8696ad650b674408a48d8d Mon Sep 17 00:00:00 2001 From: Asher Cohen Date: Tue, 23 Jun 2026 21:58:56 -0700 Subject: [PATCH 11/11] Fix CI lint, E2E migration, and Garmin integration ClickHouse setup. Accept numeric legacy table counts from ClickHouse, fix Biome import order, and bootstrap ingest.metric_stream in Garmin integration tests when CLICKHOUSE_URL is set. Co-authored-by: Cursor --- src/db/clickhouse-migrations.test.ts | 10 +++++----- .../0034_move_metric_stream_to_ingest.test.ts | 16 +++++++++++++++- .../0034_move_metric_stream_to_ingest.ts | 2 +- ...min-connect-internal-sync.integration.test.ts | 12 ++++++++++++ src/providers/garmin/sync-checkpoint.test.ts | 1 + 5 files changed, 34 insertions(+), 7 deletions(-) diff --git a/src/db/clickhouse-migrations.test.ts b/src/db/clickhouse-migrations.test.ts index 617ad8a64a..ae29e6f4fd 100644 --- a/src/db/clickhouse-migrations.test.ts +++ b/src/db/clickhouse-migrations.test.ts @@ -30,6 +30,11 @@ import * as activityUserSoftDeleteMigration from "./clickhouse-migrations/0031_a import * as dedupedActivitiesAbsentSourceLinksMigration from "./clickhouse-migrations/0032_deduped_activities_absent_source_links.ts"; import { clickHouseMigrationFileNames } from "./clickhouse-migrations/registry.ts"; import { runClickHouseMigrationStatement } from "./clickhouse-migrations/statement-runner.ts"; +import { + buildClickHouseMigrationStatements, + runClickHouseMigrations, +} from "./clickhouse-migrations.ts"; +import { replacingMergeTreeTable } from "./clickhouse-sql-helpers.ts"; function systemTablesQueryRows( queryText: string, @@ -39,11 +44,6 @@ function systemTablesQueryRows( } return [{ table_count: 1 }]; } -import { - buildClickHouseMigrationStatements, - runClickHouseMigrations, -} from "./clickhouse-migrations.ts"; -import { replacingMergeTreeTable } from "./clickhouse-sql-helpers.ts"; describe("buildClickHouseMigrationStatements", () => { it("loads ClickHouse migrations from ordered per-file modules", () => { diff --git a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts index 86867b1a92..8f4ce6e503 100644 --- a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts +++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts @@ -61,7 +61,7 @@ describe("0034_move_metric_stream_to_ingest", () => { it.each([ { label: "no rows", countRows: [] }, { label: "a non-array response", countRows: { 0: { count: "1" } } }, - { label: "a non-string count", countRows: [{ count: 1 }] }, + { label: "a negative count", countRows: [{ count: -1 }] }, { label: "a missing count field", countRows: [{ table_count: "1" }] }, { label: "a null row", countRows: [null] }, ])("rejects malformed legacy table count payloads ($label)", async ({ countRows }) => { @@ -95,6 +95,20 @@ describe("0034_move_metric_stream_to_ingest", () => { }); }); + it("copies legacy metric stream rows when ClickHouse returns a numeric count", async () => { + const migration = moveMetricStreamToIngestMigration.createMigration(); + const client = mockLegacyTableCountClient([{ count: 1 }]); + + await migration.run?.(client, "postgres://health:fixture@db:5432/health"); + + expect(client.command).toHaveBeenCalledWith({ + query: expect.stringContaining(`INSERT INTO ${METRIC_STREAM_TABLE}`), + }); + expect(client.command).toHaveBeenCalledWith({ + query: `DROP TABLE IF EXISTS ${LEGACY_METRIC_STREAM_TABLE}`, + }); + }); + it("copies legacy metric stream rows when the legacy table count is greater than zero", async () => { const migration = moveMetricStreamToIngestMigration.createMigration(); const client = mockLegacyTableCountClient([{ count: "2" }]); diff --git a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts index d9070b1f0a..9f898e36b7 100644 --- a/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts +++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts @@ -11,7 +11,7 @@ import type { ClickHouseMigration } from "./types.ts"; const legacyTableCountRowsSchema = z .array( z.object({ - count: z.string().regex(/^\d+$/), + count: z.union([z.string().regex(/^\d+$/), z.number().int().nonnegative()]), }), ) .nonempty("Expected at least one row from system.tables count query"); diff --git a/src/providers/garmin-connect-internal-sync.integration.test.ts b/src/providers/garmin-connect-internal-sync.integration.test.ts index 2ce7e71fb8..3f710e7f71 100644 --- a/src/providers/garmin-connect-internal-sync.integration.test.ts +++ b/src/providers/garmin-connect-internal-sync.integration.test.ts @@ -1,5 +1,7 @@ import { eq } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { createClickHouseClientFromEnv } from "../db/clickhouse.ts"; +import { buildClickHouseBootstrapStatementsForNativeMetricStream } from "../db/clickhouse-metric-stream-bootstrap.ts"; import { activity, dailyMetrics, oauthToken, sleepSession, userSettings } from "../db/schema.ts"; import { setupTestDatabase, type TestContext } from "../db/test-helpers.ts"; import { ensureProvider, saveTokens } from "../db/tokens.ts"; @@ -377,6 +379,16 @@ describe("GarminProvider.sync() internal Connect API (integration)", () => { beforeAll(async () => { ctx = await setupTestDatabase(); await ensureProvider(ctx.db, "garmin", "Garmin Connect"); + if (process.env.CLICKHOUSE_URL) { + const client = createClickHouseClientFromEnv(); + try { + for (const statement of buildClickHouseBootstrapStatementsForNativeMetricStream("")) { + await client.command({ query: statement }); + } + } finally { + await client.close?.(); + } + } }, 60_000); beforeEach(() => { diff --git a/src/providers/garmin/sync-checkpoint.test.ts b/src/providers/garmin/sync-checkpoint.test.ts index a71ce2a920..b200f9b9d2 100644 --- a/src/providers/garmin/sync-checkpoint.test.ts +++ b/src/providers/garmin/sync-checkpoint.test.ts @@ -14,6 +14,7 @@ const clickHouseMocks = vi.hoisted(() => { vi.mock("../../db/clickhouse.ts", () => ({ createClickHouseClientFromEnv: clickHouseMocks.createClickHouseClientFromEnv, })); + import { applyRateLimitToCheckpoint, createGarminSyncCheckpoint,