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..a0de54d0a6 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(sourcesYaml).toContain("name: ingest");
+    expect(sourcesYaml).toContain("event_time: ingested_at");
     expect(sensorScalarSampleSql).toContain("event_time='_peerdb_synced_at'");
-    expect(sensorScalarSampleSql).toContain(
-      "source('postgres_fitness', 'metric_stream_freshness')",
-    );
+    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..5aed5c4e9c 100644
--- a/analytics/models/staging/sensor_scalar_sample.sql
+++ b/analytics/models/staging/sensor_scalar_sample.sql
@@ -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..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;
-  _peerdb_is_deleted: 0 | 1;
-  _peerdb_version: number;
-}
-
 const client = createClickHouseClientFromEnv();
 
 const reader: MetricStreamClickHouseReader = {
@@ -27,22 +21,8 @@ const reader: MetricStreamClickHouseReader = {
   },
 };
 
-async function seed(rows: SeedRow[]): Promise<void> {
-  await client.insert?.({
-    table: "postgres_fitness.metric_stream",
-    format: "JSONEachRow",
-    values: rows.map((row) => ({
-      id: row.id,
-      user_id: testUserId,
-      recorded_at: row.recorded_at,
-      provider_id: row.provider_id,
-      channel: row.channel,
-      scalar: row.scalar,
-      _peerdb_is_deleted: row._peerdb_is_deleted,
-      _peerdb_synced_at: "2026-04-12 00:00:00.000",
-      _peerdb_version: row._peerdb_version,
-    })),
-  });
+async function seed(rows: MetricStreamSeedRow[]): Promise<void> {
+  await seedMetricStreamRows(client, testUserId, rows);
 }
 
 describe("HeartRateRepository (integration)", () => {
@@ -51,7 +31,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 },
     });
 
@@ -64,8 +44,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 +53,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 +63,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 +73,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 +82,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 +92,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 +101,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 +110,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,15 +119,15 @@ 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);
 
   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..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<HeartRateSourceSeries[]> {
@@ -61,10 +61,10 @@ export class HeartRateRepository {
             provider_id,
             toStartOfMinute(toDateTime(recorded_at)) AS minute_bucket,
             scalar
-          FROM postgres_fitness.metric_stream FINAL
-          WHERE user_id = {userId:String}
+          FROM ingest.metric_stream FINAL
+          WHERE user_id = {userId:UUID}
             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/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<string, unknown>[];
+  }) => Promise<unknown>;
+}
+
+export async function seedMetricStreamRows(
+  client: MetricStreamSeedClient,
+  userId: string,
+  rows: MetricStreamSeedRow[],
+): Promise<void> {
+  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 dfe658e318..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<Database, "execute" | "transaction"> = {
 
 const testUserId = "00000000-0000-0000-0000-0000000000b2";
 
-interface SeedRow {
-  id: string;
-  recorded_at: string;
-  provider_id: string;
-  channel: string;
-  scalar: number;
-  _peerdb_is_deleted: 0 | 1;
-  _peerdb_version: number;
-}
-
 const client = createClickHouseClientFromEnv();
 
 // The repository only needs a ClickHouse query runner (BodyClickHouseStore shape).
@@ -38,22 +32,8 @@ const clickHouse: BodyClickHouseStore = {
   },
 };
 
-async function seed(rows: SeedRow[]): Promise<void> {
-  await client.insert?.({
-    table: "postgres_fitness.metric_stream",
-    format: "JSONEachRow",
-    values: rows.map((row) => ({
-      id: row.id,
-      user_id: testUserId,
-      recorded_at: row.recorded_at,
-      provider_id: row.provider_id,
-      channel: row.channel,
-      scalar: row.scalar,
-      _peerdb_is_deleted: row._peerdb_is_deleted,
-      _peerdb_synced_at: "2026-04-12 00:00:00.000",
-      _peerdb_version: row._peerdb_version,
-    })),
-  });
+async function seed(rows: MetricStreamSeedRow[]): Promise<void> {
+  await seedMetricStreamRows(client, testUserId, rows);
 }
 
 const rowSchema = z.object({
@@ -72,7 +52,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([
@@ -82,8 +62,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 +71,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 +81,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 +90,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 +100,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,15 +109,15 @@ 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);
 
   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 344d1bb685..40b910ccbc 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"],
@@ -203,7 +203,6 @@ describe("ProviderDetailRepository", () => {
       expect(query.mock.calls[0]?.[2]).toStrictEqual({
         userId: "user-1",
         providerId: "withings",
-        recordId: "",
         limit: 10,
         offset: 5,
       });
@@ -226,18 +225,17 @@ describe("ProviderDetailRepository", () => {
 
       expect(query).toHaveBeenCalledTimes(1);
       // Reads the deduped Redpanda-fed mirror, not Postgres.
-      expect(query.mock.calls[0]?.[1]).toContain("FROM postgres_fitness.metric_stream");
+      expect(query.mock.calls[0]?.[1]).toContain("FROM ingest.metric_stream");
       expect(query.mock.calls[0]?.[1]).toContain(
-        "row_number() OVER (PARTITION BY id ORDER BY _peerdb_version DESC)",
+        "row_number() OVER (PARTITION BY id ORDER BY version DESC)",
       );
       expect(query.mock.calls[0]?.[1]).toContain("version_rank = 1");
-      expect(query.mock.calls[0]?.[1]).toContain("_peerdb_is_deleted = 0");
+      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({
         userId: "user-1",
         providerId: "whoop",
-        recordId: "",
         limit: 10,
         offset: 5,
       });
@@ -484,7 +482,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 139f7b4a30..85b2031d9f 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(
@@ -325,7 +325,7 @@ export class ProviderDetailRepository {
       {
         userId: this.#userId,
         providerId,
-        recordId: recordId ?? "",
+        ...(recordId ? { recordId } : {}),
         limit,
         offset,
       },
@@ -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
-   * via `row_number()` over `_peerdb_version` (avoids a full-table `FINAL` merge
+   * via `row_number()` over `version` (avoids a full-table `FINAL` merge
    * on large continuous streams such as WHOOP heart rate); recorded_at formatted
    * to ISO-8601 Z.
    */
@@ -376,15 +376,15 @@ export class ProviderDetailRepository {
             channel,
             activity_id,
             scalar,
-            _peerdb_is_deleted,
-            row_number() OVER (PARTITION BY id ORDER BY _peerdb_version DESC) AS version_rank
-          FROM postgres_fitness.metric_stream
+            is_deleted,
+            row_number() OVER (PARTITION BY id ORDER BY version DESC) AS version_rank
+          FROM ingest.metric_stream
           WHERE user_id = {userId:UUID}
             AND provider_id = {providerId:String}
             ${recordFilter}
         ) AS ranked_versions
         WHERE version_rank = 1
-          AND _peerdb_is_deleted = 0
+          AND is_deleted = 0
         ORDER BY recorded_at DESC
         LIMIT {limit:UInt32}
         OFFSET {offset:UInt32}
@@ -392,7 +392,7 @@ export class ProviderDetailRepository {
       {
         userId: this.#userId,
         providerId,
-        recordId: recordId ?? "",
+        ...(recordId ? { recordId } : {}),
         limit,
         offset,
       },
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/packages/server/src/routers/clickhouse-integration-test-helpers.ts b/packages/server/src/routers/clickhouse-integration-test-helpers.ts
index 8ab62560d2..0a48ce79c9 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?.();
   });
@@ -1595,88 +1584,64 @@ async function bootstrapClickHouseTestSchema(
   client: ClickHouseClient,
   connectionString: string,
 ): Promise<void> {
+  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() });
 }
@@ -1711,9 +1676,6 @@ async function syncClickHouseTestActivitySensorStoreWithClient(
   connectionString: string,
 ): Promise<void> {
   for (const rawTableSync of rawTableSyncs) {
-    if (rawTableSync.tableName === "metric_stream") {
-      continue;
-    }
     await client.command({
       query: `TRUNCATE TABLE postgres_fitness.${rawTableSync.tableName}`,
     });
@@ -1789,7 +1751,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 +1765,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 14fbd1471c..3b765df7c3 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"],
@@ -605,8 +605,12 @@ describe("providerDetailRouter", () => {
 
       expect(mockExecute).not.toHaveBeenCalled();
       expect(sensorQuery).toHaveBeenCalledTimes(1);
-      expect(sensorQuery.mock.calls[0]?.[1]).toContain("postgres_fitness.metric_stream");
+      expect(sensorQuery.mock.calls[0]?.[1]).toContain("FROM ingest.metric_stream");
+      expect(sensorQuery.mock.calls[0]?.[1]).toContain(
+        "row_number() OVER (PARTITION BY id ORDER BY version DESC)",
+      );
       expect(sensorQuery.mock.calls[0]?.[1]).toContain("version_rank = 1");
+      expect(sensorQuery.mock.calls[0]?.[1]).toContain("is_deleted = 0");
       expect(result.rows).toHaveLength(1);
       expect(result.rows[0]?.channel).toBe("heart_rate");
     });
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<void> {
   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..d49edd1562 100644
--- a/scripts/seed-review-clickhouse.test.ts
+++ b/scripts/seed-review-clickhouse.test.ts
@@ -20,9 +20,13 @@ 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("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 164d280200..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(", ");
 
@@ -151,7 +171,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",
@@ -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-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..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,37 +12,16 @@ import {
   buildBodyMeasurementSampleProjectionStatements,
 } from "./clickhouse-read-models.ts";
 import { buildRestingHeartRateSleepWindowTableSql } from "./clickhouse-resting-heart-rate.ts";
-import {
-  peerDbMetadataColumnAlterStatements,
-  peerDbMetadataColumnDefinitions,
-  replacingMergeTreeTable,
-  standardViewHeader,
-} from "./clickhouse-sql-helpers.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..ae29e6f4fd 100644
--- a/src/db/clickhouse-migrations.test.ts
+++ b/src/db/clickhouse-migrations.test.ts
@@ -36,6 +36,15 @@ import {
 } from "./clickhouse-migrations.ts";
 import { replacingMergeTreeTable } from "./clickhouse-sql-helpers.ts";
 
+function systemTablesQueryRows(
+  queryText: string,
+): Array<{ count: string } | { table_count: number }> {
+  if (queryText.includes("AS count")) {
+    return [{ count: "1" }];
+  }
+  return [{ table_count: 1 }];
+}
+
 describe("buildClickHouseMigrationStatements", () => {
   it("loads ClickHouse migrations from ordered per-file modules", () => {
     expect(clickHouseMigrationFileNames).toEqual([
@@ -72,6 +81,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 +272,19 @@ 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).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");
@@ -402,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 {
@@ -455,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" }]) };
@@ -471,7 +476,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 +496,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(
@@ -543,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 }]) };
@@ -617,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")
@@ -642,7 +647,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 +659,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')",
@@ -727,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") ||
@@ -747,7 +752,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:
@@ -784,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")
@@ -808,7 +813,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"),
       }),
     );
   });
@@ -822,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")
@@ -844,7 +849,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({
@@ -880,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")
@@ -904,17 +909,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"),
       }),
     );
   });
@@ -946,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")
@@ -972,12 +977,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"),
       }),
     );
   });
@@ -1018,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")
@@ -1033,7 +1038,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')",
@@ -1070,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")
@@ -1085,7 +1090,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')",
@@ -1108,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")
@@ -1158,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" }]) };
@@ -1185,12 +1190,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 +1205,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"),
       }),
     );
   });
@@ -1212,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")
@@ -1262,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")
@@ -1277,7 +1282,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')",
@@ -1329,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")
@@ -1353,7 +1358,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')",
@@ -1372,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")
@@ -1396,7 +1401,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"),
       }),
     );
   });
@@ -1427,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")
@@ -1477,11 +1482,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.test.ts b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts
new file mode 100644
index 0000000000..8f4ce6e503
--- /dev/null
+++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.test.ts
@@ -0,0 +1,125 @@
+import { describe, expect, it, vi } from "vitest";
+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 {
+  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.each([
+    { label: "no rows", countRows: [] },
+    { label: "a non-array response", countRows: { 0: { 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 }) => {
+    const migration = moveMetricStreamToIngestMigration.createMigration();
+    const client = mockLegacyTableCountClient(countRows);
+
+    await expect(
+      migration.run?.(client, "postgres://health:fixture@db:5432/health"),
+    ).rejects.toThrow();
+    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 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" }]);
+
+    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}`,
+    });
+  });
+});
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..9f898e36b7
--- /dev/null
+++ b/src/db/clickhouse-migrations/0034_move_metric_stream_to_ingest.ts
@@ -0,0 +1,78 @@
+import { z } from "zod";
+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";
+
+const legacyTableCountRowsSchema = z
+  .array(
+    z.object({
+      count: z.union([z.string().regex(/^\d+$/), z.number().int().nonnegative()]),
+    }),
+  )
+  .nonempty("Expected at least one row from system.tables count query");
+
+function parseLegacyTableCount(rows: unknown): number {
+  const [firstRow] = legacyTableCountRowsSchema.parse(rows);
+  return Number(firstRow.count);
+}
+
+async function legacyMetricStreamTableExists(client: ClickHouseCommandClient): Promise<boolean> {
+  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
+      WHERE database = 'postgres_fitness'
+        AND name = 'metric_stream'`,
+    format: "JSONEachRow",
+  });
+  return parseLegacyTableCount(await result.json()) > 0;
+}
+
+async function copyLegacyMetricStreamToIngest(client: ClickHouseCommandClient): Promise<void> {
+  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..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,
@@ -190,10 +191,7 @@ export async function rebuildMetricStreamLocationPoint(
       client,
       "DROP TABLE IF EXISTS analytics.metric_stream_backfill_chunks",
     );
-    await runClickHouseMigrationStatement(
-      client,
-      "DROP TABLE IF EXISTS postgres_fitness.metric_stream",
-    );
+    await runClickHouseMigrationStatement(client, `DROP TABLE IF EXISTS ${METRIC_STREAM_TABLE}`);
   }
 
   for (const statement of buildClickHouseBootstrapStatements(postgresConnectionString)) {
@@ -220,7 +218,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 +268,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<MigrationCountRow>({
-    query: `SELECT count() AS migration_count
+  for (const database of [INGEST_DATABASE, "postgres_fitness"]) {
+    const result = await client.query<MigrationCountRow>({
+      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(
@@ -306,8 +309,7 @@ async function shouldReplaceMetricStreamTable(client: ClickHouseCommandClient):
     throw new Error("ClickHouse migrations require a query-capable client");
   }
   const result = await client.query<ClickHouseDatabaseEngineRow>({
-    query:
-      "SELECT engine FROM system.tables WHERE database = 'postgres_fitness' 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();
@@ -341,8 +343,8 @@ async function backfillNativeMetricStream(
   client: ClickHouseCommandClient,
   postgresConnectionString: string,
 ): Promise<void> {
-  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 +528,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 +584,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<void> {
   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-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}
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..b20602ee54 100644
--- a/src/db/clickhouse.ts
+++ b/src/db/clickhouse.ts
@@ -1,4 +1,5 @@
 import { createClient } from "@clickhouse/client";
+import { METRIC_STREAM_TABLE } from "../metric-stream/clickhouse-table.ts";
 import { buildClickHouseBootstrapStatementsForNativeMetricStream } from "./clickhouse-metric-stream-bootstrap.ts";
 
 export interface ClickHouseCommandClient {
@@ -111,12 +112,12 @@ export function createClickHouseClientFromEnv(
 }
 
 export async function bootstrapClickHouseFromEnv(client: ClickHouseCommandClient): Promise<void> {
-  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<void> {
   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..4d314fecc4 100644
--- a/src/metric-stream/clickhouse-sink.test.ts
+++ b/src/metric-stream/clickhouse-sink.test.ts
@@ -5,6 +5,7 @@ import {
   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";
 
@@ -31,6 +32,14 @@ const imuEvent = {
   channel: "imu",
 } satisfies MetricStreamEventV1;
 
+function firstCommandQuery(command: ReturnType<typeof vi.fn>): 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 call.query;
+}
+
 afterEach(() => {
   vi.doUnmock("../db/clickhouse.ts");
   vi.doUnmock("./redpanda-consumer.ts");
@@ -38,14 +47,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 +65,7 @@ describe("insertMetricStreamEventsIntoClickHouse", () => {
           external_id: "hk:heart-rate-1",
           device_id: "Apple Watch",
           scalar: 72,
-          _peerdb_is_deleted: 0,
+          is_deleted: 0,
         }),
       ],
     });
@@ -94,7 +103,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 +148,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(firstCommandQuery(command))).toContain("1 AS is_deleted");
+    expect(String(firstCommandQuery(command))).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 +186,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 +200,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 +218,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 +230,14 @@ describe("applyMetricStreamEventsToClickHouse", () => {
         recorded_at_end: "2026-03-02T00:00:00.000Z",
       },
     });
+    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}");
+    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<string, unknown> }): Promise<unknown>;
   insert(options: {
-    table: "postgres_fitness.metric_stream";
+    table: typeof METRIC_STREAM_TABLE;
     values: readonly ClickHouseMetricStreamRow[];
     format: "JSONEachRow";
     clickhouse_settings?: Record<string, string | number | boolean>;
@@ -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<string, unknown> = {
-    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-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.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..b200f9b9d2 100644
--- a/src/providers/garmin/sync-checkpoint.test.ts
+++ b/src/providers/garmin/sync-checkpoint.test.ts
@@ -1,5 +1,20 @@
-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 +24,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(),
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 a1b5844292..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,19 +139,18 @@ async function listGarminDatesWithMetricStreamChannel(
     return new Set();
   }
 
-  const { createClickHouseClientFromEnv } = await import("../../db/clickhouse.ts");
   const client = createClickHouseClientFromEnv();
   try {
     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: {
@@ -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?.();
   }
diff --git a/stryker.ci.config.json b/stryker.ci.config.json
index 87a381b155..2eeada5cd2 100644
--- a/stryker.ci.config.json
+++ b/stryker.ci.config.json
@@ -45,15 +45,18 @@
   ],
   "ignorePatterns": [
     "dist/**",
-    "**/skills/stripe-projects",
+    ".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 15a6d383bb..323a2d630a 100644
--- a/stryker.config.json
+++ b/stryker.config.json
@@ -46,15 +46,18 @@
   ],
   "ignorePatterns": [
     "dist/**",
-    "**/skills/stripe-projects",
+    ".adal/**",
+    ".aider-desk/**",
     ".antigravitycli/**",
+    ".augment/**",
     ".codegraph/**",
     ".codex/**",
     ".claude/**",
     ".context/**",
     ".gemini/**",
     ".kilocode/**",
-    ".agents/**"
+    ".agents/**",
+    "**/skills/stripe-projects"
   ],
   "mutator": {
     "excludedMutations": ["StringLiteral"]