Fix metric stream sink deletes and move table to ingest

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

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

analytics/models/read_models/read_model_microbatch.sql.test.ts

@@ -259,20 +259,15 @@ describe("production analytics read-model build", () => {
 
     expect(sourcesYaml).toContain("name: metric_stream_freshness");
     expect(sourcesYaml).toContain("identifier: metric_stream");
-    expect(sourcesYaml).toContain("event_time: _peerdb_synced_at");
-    expect(sensorScalarSampleSql).toContain("event_time='_peerdb_synced_at'");
-    expect(sensorScalarSampleSql).toContain(
-      "source('postgres_fitness', 'metric_stream_freshness')",
-    );
+    expect(sourcesYaml).toContain("name: ingest");
+    expect(sourcesYaml).toContain("event_time: ingested_at");
+    expect(sensorScalarSampleSql).toContain("event_time='ingested_at'");
+    expect(sensorScalarSampleSql).toContain("source('ingest', 'metric_stream_freshness')");
     expect(dedupedSensorSql).toContain("event_time='refreshed_at'");
     expect(dedupedSensorSql).toContain("max(samples._peerdb_synced_at) AS source_refreshed_at");
     expect(dedupedSensorSql).toContain("source_refreshed_at AS refreshed_at");
-    expect(activityLocationSampleSql).toContain(
-      "source('postgres_fitness', 'metric_stream_freshness')",
-    );
-    expect(activityLocationSampleSql).not.toContain(
-      "source('postgres_fitness', 'metric_stream')",
-    );
+    expect(activityLocationSampleSql).toContain("source('ingest', 'metric_stream_freshness')");
+    expect(activityLocationSampleSql).not.toContain("source('postgres_fitness', 'metric_stream')");
   });
 
   it("materializes activity location membership as a microbatch intermediary", () => {
@@ -282,11 +277,11 @@ describe("production analytics read-model build", () => {
     expect(sql).toContain("incremental_strategy='microbatch'");
     expect(sql).toContain("event_time='refreshed_at'");
     expect(sql).toContain("lookback=3");
-    expect(sql).toContain("source('postgres_fitness', 'metric_stream_freshness')");
+    expect(sql).toContain("source('ingest', 'metric_stream_freshness')");
     expect(sql).toContain("ref('deduped_activity_members')");
     expect(sql).not.toContain("source('analytics', 'v_activity_members')");
     expect(sql).toContain("channel = 'location'");
-    expect(sql).toContain("argMax(point, _peerdb_version) AS point");
+    expect(sql).toContain("argMax(point, version) AS point");
     expect(sql).toContain("toString(point) AS point_text");
     expect(sql).toContain("startsWith(location_rows.point_text, '{')");
     expect(sql).toContain("JSONExtract(location_rows.point_text, 'coordinates', 'Array(Float64)')[2]");
@@ -319,7 +314,7 @@ describe("production analytics read-model build", () => {
     );
     expect(activitySensorSampleSql).toContain("source_refreshed_at AS refreshed_at");
     expect(activityLocationSampleSql).toContain(
-      "greatest(location_rows._peerdb_synced_at, activity_members.source_synced_at) AS source_refreshed_at",
+      "greatest(location_rows.ingested_at, activity_members.source_synced_at) AS source_refreshed_at",
     );
     expect(activityLocationSampleSql).toContain("source_refreshed_at AS refreshed_at");
     expect(activityLocationSampleSql).not.toContain("now64(9) AS refreshed_at");

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

analytics/models/staging/sensor_scalar_sample.sql

@@ -2,7 +2,7 @@
     materialized='incremental',
     incremental_strategy='microbatch',
     unique_key='id',
-    event_time='_peerdb_synced_at',
+    event_time='ingested_at',
     begin='2026-01-01',
     batch_size='day',
     lookback=3,
@@ -17,8 +17,8 @@
 
 WITH metric_stream_versions AS (
     SELECT *
-    FROM {{ source('postgres_fitness', 'metric_stream_freshness') }}
-    WHERE (scalar IS NOT null OR _peerdb_is_deleted = 1)
+    FROM {{ source('ingest', 'metric_stream_freshness') }}
+    WHERE (scalar IS NOT null OR is_deleted = 1)
         AND channel IN (
             'heart_rate',
             'power',
@@ -41,15 +41,15 @@ WITH metric_stream_versions AS (
 metric_stream_rows AS (
     SELECT
         id,
-        argMax(user_id, _peerdb_version) AS user_id,
-        argMax(recorded_at, _peerdb_version) AS recorded_at,
-        argMax(channel, _peerdb_version) AS channel,
-        argMax(provider_id, _peerdb_version) AS provider_id,
-        argMax(device_id, _peerdb_version) AS device_id,
-        coalesce(argMax(scalar, _peerdb_version), 0) AS scalar,
-        argMax(_peerdb_synced_at, _peerdb_version) AS _peerdb_synced_at,
-        argMax(_peerdb_is_deleted, _peerdb_version) AS _peerdb_is_deleted,
-        max(_peerdb_version) AS source_peerdb_version
+        argMax(user_id, version) AS user_id,
+        argMax(recorded_at, version) AS recorded_at,
+        argMax(channel, version) AS channel,
+        argMax(provider_id, version) AS provider_id,
+        argMax(device_id, version) AS device_id,
+        coalesce(argMax(scalar, version), 0) AS scalar,
+        argMax(ingested_at, version) AS ingested_at,
+        argMax(is_deleted, version) AS is_deleted,
+        max(version) AS source_version
     FROM metric_stream_versions
     GROUP BY id
 ),
@@ -113,9 +113,9 @@ SELECT
         active_sensor_provider_priority.priority,
         1000
     ) AS provider_priority,
-    metric_stream_rows._peerdb_synced_at AS _peerdb_synced_at,
-    metric_stream_rows._peerdb_is_deleted AS _peerdb_is_deleted,
-    metric_stream_rows.source_peerdb_version AS _peerdb_version
+    metric_stream_rows.ingested_at AS _peerdb_synced_at,
+    metric_stream_rows.is_deleted AS _peerdb_is_deleted,
+    metric_stream_rows.source_version AS _peerdb_version
 FROM metric_stream_rows
 LEFT JOIN active_sensor_provider_priority
     ON active_sensor_provider_priority.provider_id = metric_stream_rows.provider_id

packages/server/src/repositories/heart-rate-repository.integration.test.ts

@@ -29,7 +29,7 @@ const reader: MetricStreamClickHouseReader = {
 
 async function seed(rows: SeedRow[]): Promise<void> {
   await client.insert?.({
-    table: "postgres_fitness.metric_stream",
+    table: "ingest.metric_stream",
     format: "JSONEachRow",
     values: rows.map((row) => ({
       id: row.id,
@@ -51,7 +51,7 @@ describe("HeartRateRepository (integration)", () => {
       await client.command({ query: statement });
     }
     await client.command({
-      query: "DELETE FROM postgres_fitness.metric_stream WHERE user_id = {userId:UUID}",
+      query: "DELETE FROM ingest.metric_stream WHERE user_id = {userId:UUID}",
       query_params: { userId: testUserId },
     });
 
@@ -147,7 +147,7 @@ describe("HeartRateRepository (integration)", () => {
 
   afterAll(async () => {
     await client.command({
-      query: "DELETE FROM postgres_fitness.metric_stream WHERE user_id = {userId:UUID}",
+      query: "DELETE FROM ingest.metric_stream WHERE user_id = {userId:UUID}",
       query_params: { userId: testUserId },
     });
     await client.close?.();

packages/server/src/repositories/heart-rate-repository.ts

@@ -61,10 +61,10 @@ export class HeartRateRepository {
             provider_id,
             toStartOfMinute(toDateTime(recorded_at)) AS minute_bucket,
             scalar
-          FROM postgres_fitness.metric_stream FINAL
+          FROM ingest.metric_stream FINAL
           WHERE user_id = {userId:String}
             AND channel = 'heart_rate'
-            AND _peerdb_is_deleted = 0
+            AND is_deleted = 0
             AND scalar > 0
             -- Range over the raw recorded_at so the ORDER BY key can prune,
             -- instead of wrapping it in toDate(). Bounds are the user's local

packages/server/src/repositories/provider-detail-repository.integration.test.ts

@@ -40,7 +40,7 @@ const clickHouse: BodyClickHouseStore = {
 
 async function seed(rows: SeedRow[]): Promise<void> {
   await client.insert?.({
-    table: "postgres_fitness.metric_stream",
+    table: "ingest.metric_stream",
     format: "JSONEachRow",
     values: rows.map((row) => ({
       id: row.id,
@@ -72,7 +72,7 @@ describe("ProviderDetailRepository metric stream (integration)", () => {
       await client.command({ query: statement });
     }
     await client.command({
-      query: "DELETE FROM postgres_fitness.metric_stream WHERE user_id = {userId:UUID}",
+      query: "DELETE FROM ingest.metric_stream WHERE user_id = {userId:UUID}",
       query_params: { userId: testUserId },
     });
     await seed([
@@ -137,7 +137,7 @@ describe("ProviderDetailRepository metric stream (integration)", () => {
 
   afterAll(async () => {
     await client.command({
-      query: "DELETE FROM postgres_fitness.metric_stream WHERE user_id = {userId:UUID}",
+      query: "DELETE FROM ingest.metric_stream WHERE user_id = {userId:UUID}",
       query_params: { userId: testUserId },
     });
     await client.close?.();

packages/server/src/repositories/provider-detail-repository.test.ts

@@ -19,7 +19,7 @@ describe("tableInfo", () => {
     ["bodyMeasurements", "analytics.v_body_measurement", "recorded_at", "id"],
     ["foodEntries", "fitness.food_entry", "date", "id"],
     ["healthEvents", "fitness.health_event", "start_date", "id"],
-    ["metricStream", "postgres_fitness.metric_stream", "recorded_at", "id"],
+    ["metricStream", "ingest.metric_stream", "recorded_at", "id"],
     ["nutritionDaily", "fitness.v_nutrition_daily", "date", "date"],
     ["labPanels", "fitness.lab_panel", "recorded_at", "id"],
     ["labResults", "fitness.lab_result", "recorded_at", "id"],
@@ -226,8 +226,8 @@ describe("ProviderDetailRepository", () => {
 
       expect(query).toHaveBeenCalledTimes(1);
       // Reads the deduped Redpanda-fed mirror, not Postgres.
-      expect(query.mock.calls[0]?.[1]).toContain("postgres_fitness.metric_stream FINAL");
-      expect(query.mock.calls[0]?.[1]).toContain("_peerdb_is_deleted = 0");
+      expect(query.mock.calls[0]?.[1]).toContain("ingest.metric_stream FINAL");
+      expect(query.mock.calls[0]?.[1]).toContain("is_deleted = 0");
       // recorded_at is rendered in UTC so the literal 'Z' suffix is accurate.
       expect(query.mock.calls[0]?.[1]).toContain("'%Y-%m-%dT%H:%i:%S.%fZ', 'UTC'");
       expect(query.mock.calls[0]?.[2]).toStrictEqual({
@@ -480,7 +480,7 @@ describe("ProviderDetailRepository", () => {
         idColumn: "id",
       });
       expect(tableInfo("metricStream")).toStrictEqual({
-        table: "postgres_fitness.metric_stream",
+        table: "ingest.metric_stream",
         orderColumn: "recorded_at",
         idColumn: "id",
       });

packages/server/src/repositories/provider-detail-repository.ts

@@ -45,7 +45,7 @@ export function tableInfo(dataType: DataType): {
       return { table: "fitness.health_event", orderColumn: "start_date", idColumn: "id" };
     case "metricStream":
       return {
-        table: "postgres_fitness.metric_stream",
+        table: "ingest.metric_stream",
         orderColumn: "recorded_at",
         idColumn: "id",
       };
@@ -231,7 +231,7 @@ export class ProviderDetailRepository {
   readonly #userId: string;
   // ClickHouse query runner. Serves the read-models that have moved off
   // Postgres: body measurements (analytics.v_body_measurement) and the
-  // Redpanda-fed metric stream (postgres_fitness.metric_stream).
+  // Redpanda-fed metric stream (ingest.metric_stream).
   readonly #clickHouse: BodyClickHouseStore | undefined;
 
   constructor(
@@ -335,7 +335,7 @@ export class ProviderDetailRepository {
   /**
    * Raw metric-stream rows for a provider, read from the Redpanda-fed ClickHouse
    * mirror (Postgres `fitness.metric_stream` is retired). Version-deduplicated
-   * (`FINAL` + `_peerdb_is_deleted = 0`); recorded_at formatted to ISO-8601 Z.
+   * (`FINAL` + `is_deleted = 0`); recorded_at formatted to ISO-8601 Z.
    */
   async #queryMetricStreamRecords(
     providerId: string,
@@ -363,10 +363,10 @@ export class ProviderDetailRepository {
           channel,
           activity_id,
           scalar
-        FROM postgres_fitness.metric_stream FINAL
+        FROM ingest.metric_stream FINAL
         WHERE user_id = {userId:UUID}
           AND provider_id = {providerId:String}
-          AND _peerdb_is_deleted = 0
+          AND is_deleted = 0
           ${recordFilter}
         ORDER BY recorded_at DESC
         LIMIT {limit:UInt32}