Remove sensor priority tables from CDC and clean up legacy validation mirror

README.md

@@ -45,7 +45,7 @@ If you are starting cold and do not want to hunt through agent notes, begin here
 └─────────────┘                            └───────────┘
 ```
 
-Each data source is a **provider plugin** that implements a simple interface. The sync runner orchestrates all enabled providers. Raw data lands in a `fitness` Postgres schema. ClickHouse maintains stored analytics read models for heavy activity stream reads, with scalar `metric_stream` rows copied from Postgres through the tracked ClickHouse migration path. PeerDB is deployed as the internal CDC path from Postgres to ClickHouse, with the first mirror landing in `peerdb.metric_stream` for validation before production analytics switch to it. Derived rows are not written back to Postgres. The web dashboard provides sync controls, provider health monitoring, insights, and data exploration. A companion iOS app (Expo + React Native) provides native HealthKit integration and on-the-go access. Nutrition logging on web and iOS supports natural-language AI meal input that can split one message into multiple food items. Long-running sync jobs are processed by BullMQ workers backed by Redis. In production, the `worker` container registers repeatable scheduled sync jobs in BullMQ; the `sync` mode remains available for manual one-shot runs.
+Each data source is a **provider plugin** that implements a simple interface. The sync runner orchestrates all enabled providers. Raw data lands in a `fitness` Postgres schema. ClickHouse maintains stored analytics read models for heavy activity stream reads, with scalar `metric_stream` rows copied from Postgres through the tracked ClickHouse migration path. PeerDB is deployed as the internal CDC path from Postgres to ClickHouse, replicating `metric_stream` and raw fitness tables into `postgres_fitness.*` for analytics read models. Derived rows are not written back to Postgres. The web dashboard provides sync controls, provider health monitoring, insights, and data exploration. A companion iOS app (Expo + React Native) provides native HealthKit integration and on-the-go access. Nutrition logging on web and iOS supports natural-language AI meal input that can split one message into multiple food items. Long-running sync jobs are processed by BullMQ workers backed by Redis. In production, the `worker` container registers repeatable scheduled sync jobs in BullMQ; the `sync` mode remains available for manual one-shot runs.
 
 ## Quick Start
 

deploy/README.md

@@ -10,7 +10,7 @@ Dofek is deployed as a **single-node Docker Swarm** stack on **Hetzner Cloud** (
 - **Storage**:
   - **PostgreSQL**: Managed via TimescaleDB with PostGIS enabled for geospatial metric data.
   - **ClickHouse**: Runs in the swarm as the stored analytics read-model service for heavy activity stream reads. The raw `metric_stream` copy is managed through tracked ClickHouse migrations and chunk-range backfill. See [docs/clickhouse-metric-stream.md](../docs/clickhouse-metric-stream.md).
-  - **PeerDB**: Runs internally in the swarm as the Postgres-to-ClickHouse CDC service. Its first `metric_stream` mirror writes to `peerdb.metric_stream`; production analytics keep reading the migrated native copy until that initial snapshot is verified.
+  - **PeerDB**: Runs internally in the swarm as the Postgres-to-ClickHouse CDC service. It replicates `metric_stream` and raw fitness tables into `postgres_fitness.*` for analytics read models.
   - **Volume**: Terraform provisions a Hetzner Block Storage volume (`data_volume_size_gb`, default `100GB`) attached with `automount=true`.
   - **Stable mount alias**: Terraform maintains `/mnt/dofek-data` as a symlink to the attached Hetzner volume mount path (`/mnt/HC_Volume_<id>`).
   - **DB data path**: The `db` service bind-mounts Postgres data to `/mnt/dofek-data/postgres`.

docs/clickhouse-metric-stream.md

@@ -5,19 +5,16 @@ keeps a native `MergeTree` copy of the raw stream and backfills it from
 Postgres by real Timescale chunk ranges. We do not use ClickHouse
 `MaterializedPostgreSQL` for this hypertable because the hypertable root does
 not contain the physical rows; the data live in Timescale chunk tables.
-PeerDB is the CDC path for ongoing Postgres-to-ClickHouse replication. It now
-writes to two targets: `peerdb.metric_stream` for validation and
-`postgres_fitness.metric_stream` for the active analytics source.
+PeerDB is the CDC path for ongoing Postgres-to-ClickHouse replication.
+`dofek_metric_stream_analytics` replicates into
+`postgres_fitness.metric_stream`, the active analytics source.
 
 ```text
 Postgres/Timescale fitness.metric_stream
         |                         |
-        | chunk-range native backfill | peerdb peer
-        |                         |
-        |                         |  PeerDB CDC mirrors
-        |                         |                |
-        |                         |                +--> peerdb.metric_stream (validation target)
-        v                         +----------------+
+        | chunk-range native backfill | PeerDB CDC mirror
+        |                         |   (dofek_metric_stream_analytics)
+        v                         v
 ClickHouse postgres_fitness.metric_stream
         |
         | refreshable materialized view, 15-minute cadence
@@ -103,7 +100,6 @@ ClickHouse migrations create and update the databases and read models:
   `analytics.deduped_location` (and GPS-derived fields in
   `analytics.activity_summary`) stop updating for new data until a replacement
   geometry replication strategy is in place.
-- `peerdb.metric_stream`: the PeerDB CDC validation target.
 - `postgres_fitness`: app-managed native ClickHouse raw mirrors with PeerDB CDC
   metadata columns. Besides the activity/sleep/body/daily/metric stream
   analytics sources, this includes provider inventory mirrors for `food_entry`,
@@ -149,7 +145,17 @@ loads `src/db/peerdb/metric-stream-cdc.sql`, substitutes deployment
 connection values, and applies the declarative PeerDB peer and mirror
 definition. Provider inventory tables are mirrored by
 `dofek_provider_inventory_raw_analytics` so existing raw analytics mirrors do
-not need to be rebuilt when inventory coverage expands. The mirrors use a
+not need to be rebuilt when inventory coverage expands.
+
+**Mirror reconciliation**: The deploy CDC setup checks whether each mirror's
+table list matches the expected mapping in `rawAnalyticsMirrorTableMappings`.
+If any expected table is missing from an existing mirror's config, the setup
+drops and recreates the entire mirror, triggering a full initial snapshot of
+all tables in that mirror. This is destructive on a resource-constrained
+server — only add tables to a mirror mapping if a ClickHouse read model
+actually consumes them.
+
+The mirrors use a
 dedicated publication name, exclude `device_id`, `source_type`, `vector`,
 `point`, and `metadata` from the metric stream mirrors, and enable soft deletes
 so delete events are represented in ClickHouse.

src/db/clickhouse-cdc.test.ts

@@ -58,8 +58,6 @@ describe("PeerDB ClickHouse CDC setup", () => {
     "provider",
     "provider_priority",
     "device_priority",
-    "sensor_provider_priority",
-    "sensor_device_priority",
     "user_profile",
   ];
   const originalDatabaseUrl = process.env.DATABASE_URL;
@@ -172,7 +170,6 @@ describe("PeerDB ClickHouse CDC setup", () => {
       },
       templateSql: "host = {{POSTGRES_HOST}}, password = {{POSTGRES_CREDENTIAL}}",
       templateValues: {
-        clickHouseDatabase: "peerdb",
         clickHouseHost: "clickhouse",
         clickHouseCredential: "clickhouse-fixture",
         clickHousePort: 9000,
@@ -185,7 +182,7 @@ describe("PeerDB ClickHouse CDC setup", () => {
       },
     });
 
-    expect(clickHouseCommands).toContain("CREATE DATABASE IF NOT EXISTS peerdb");
+    expect(clickHouseCommands).not.toContain("CREATE DATABASE IF NOT EXISTS peerdb");
     expect(clickHouseCommands).toContain(
       "ALTER TABLE postgres_fitness.metric_stream ADD COLUMN IF NOT EXISTS _peerdb_synced_at DateTime64(9) DEFAULT now()",
     );
@@ -219,7 +216,6 @@ describe("PeerDB ClickHouse CDC setup", () => {
         },
         templateSql: "missing = {{MISSING_VALUE}}",
         templateValues: {
-          clickHouseDatabase: "peerdb",
           clickHouseHost: "clickhouse",
           clickHouseCredential: "clickhouse-fixture",
           clickHousePort: 9000,
@@ -259,7 +255,6 @@ describe("PeerDB ClickHouse CDC setup", () => {
       },
       templateSql,
       templateValues: {
-        clickHouseDatabase: "peerdb",
         clickHouseHost: "clickhouse",
         clickHouseCredential: "clickhouse-fixture",
         clickHousePort: 9000,
@@ -272,34 +267,26 @@ describe("PeerDB ClickHouse CDC setup", () => {
       },
     });
 
-    expect(peerDbQueries).toHaveLength(7);
+    expect(peerDbQueries).toHaveLength(5);
     expect(peerDbQueries[0]).toContain("CREATE PEER IF NOT EXISTS dofek_postgres");
-    expect(peerDbQueries[1]).toContain("CREATE PEER IF NOT EXISTS dofek_clickhouse");
-    expect(peerDbQueries[2]).toContain(
+    expect(peerDbQueries[1]).toContain(
       "CREATE PEER IF NOT EXISTS dofek_clickhouse_postgres_fitness",
     );
-    expect(peerDbQueries[3]).toContain("CREATE MIRROR IF NOT EXISTS dofek_metric_stream_cdc");
-    expect(peerDbQueries[4]).toContain("CREATE MIRROR IF NOT EXISTS dofek_metric_stream_analytics");
-    expect(peerDbQueries[5]).toContain("CREATE MIRROR IF NOT EXISTS dofek_fitness_raw_analytics");
-    expect(peerDbQueries[6]).toContain(
+    expect(peerDbQueries[2]).toContain("CREATE MIRROR IF NOT EXISTS dofek_metric_stream_analytics");
+    expect(peerDbQueries[3]).toContain("CREATE MIRROR IF NOT EXISTS dofek_fitness_raw_analytics");
+    expect(peerDbQueries[4]).toContain(
       "CREATE MIRROR IF NOT EXISTS dofek_provider_inventory_raw_analytics",
     );
-    expect(peerDbQueries[1]).toContain("database = 'peerdb'");
-    expect(peerDbQueries[2]).toContain("database = 'postgres_fitness'");
-    expect(peerDbQueries[3]).toContain("TO dofek_clickhouse");
-    expect(peerDbQueries[3]).toContain("dofek_metric_stream_cdc");
-    expect(peerDbQueries[3]).toContain(
+    expect(peerDbQueries[1]).toContain("database = 'postgres_fitness'");
+    expect(peerDbQueries[2]).toContain("TO dofek_clickhouse_postgres_fitness");
+    expect(peerDbQueries[2]).toContain(
       "exclude: [device_id, source_type, vector, point, metadata]",
     );
+    expect(peerDbQueries[2]).not.toContain("latitude");
+    expect(peerDbQueries[2]).not.toContain("longitude");
+    expect(peerDbQueries[3]).toContain("TO dofek_clickhouse_postgres_fitness");
     expect(peerDbQueries[4]).toContain("TO dofek_clickhouse_postgres_fitness");
-    expect(peerDbQueries[4]).toContain(
-      "exclude: [device_id, source_type, vector, point, metadata]",
-    );
-    expect(peerDbQueries[4]).not.toContain("latitude");
-    expect(peerDbQueries[4]).not.toContain("longitude");
-    expect(peerDbQueries[5]).toContain("TO dofek_clickhouse_postgres_fitness");
-    expect(peerDbQueries[6]).toContain("TO dofek_clickhouse_postgres_fitness");
-    const rawMirrorSql = `${peerDbQueries[5]}\n${peerDbQueries[6]}`;
+    const rawMirrorSql = `${peerDbQueries[3]}\n${peerDbQueries[4]}`;
     for (const rawAnalyticsTable of rawAnalyticsTables) {
       expect(rawMirrorSql).toContain(`from: fitness.${rawAnalyticsTable}`);
       expect(rawMirrorSql).toContain(`to: ${rawAnalyticsTable}`);
@@ -335,7 +322,6 @@ describe("PeerDB ClickHouse CDC setup", () => {
       },
       templateSql,
       templateValues: {
-        clickHouseDatabase: "peerdb",
         clickHouseHost: "clickhouse",
         clickHouseCredential: "clickhouse-fixture",
         clickHousePort: 9000,
@@ -373,7 +359,7 @@ describe("PeerDB ClickHouse CDC setup", () => {
                 {
                   name: "dofek_fitness_raw_analytics",
                   raw_analytics_mirror_config:
-                    "activity sleep_session sleep_stage daily_metrics provider provider_priority device_priority user_profile",
+                    "activity sleep_session sleep_stage daily_metrics provider provider_priority device_priority",
                 },
                 {
                   name: "dofek_provider_inventory_raw_analytics",
@@ -395,7 +381,6 @@ describe("PeerDB ClickHouse CDC setup", () => {
       },
       templateSql,
       templateValues: {
-        clickHouseDatabase: "peerdb",
         clickHouseHost: "clickhouse",
         clickHouseCredential: "clickhouse-fixture",
         clickHousePort: 9000,
@@ -436,7 +421,6 @@ describe("PeerDB ClickHouse CDC setup", () => {
       },
       templateSql: "first {{POSTGRES_CREDENTIAL}}; second;",
       templateValues: {
-        clickHouseDatabase: "peerdb",
         clickHouseHost: "clickhouse",
         clickHouseCredential: "clickhouse-fixture",
         clickHousePort: 9000,
@@ -491,7 +475,7 @@ describe("PeerDB ClickHouse CDC setup", () => {
     });
     expect(clickHouseClientMocks.createClickHouseClientFromEnv).toHaveBeenCalled();
     expect(peerDbClientMocks.connect).toHaveBeenCalledTimes(2);
-    expect(clickHouseClientMocks.command).toHaveBeenCalledWith({
+    expect(clickHouseClientMocks.command).not.toHaveBeenCalledWith({
       query: "CREATE DATABASE IF NOT EXISTS peerdb",
     });
     expect(clickHouseClientMocks.command).toHaveBeenCalledWith({
@@ -512,35 +496,28 @@ describe("PeerDB ClickHouse CDC setup", () => {
     const peerDbQueries = peerDbClientMocks.query.mock.calls
       .map(([queryText]) => String(queryText))
       .filter((queryText) => !isPeerDbMirrorReconciliationQuery(queryText));
-    expect(peerDbQueries).toHaveLength(9);
+    expect(peerDbQueries).toHaveLength(7);
     expect(peerDbQueries[0]).toContain("peerdb_metric_stream_publication");
     expect(peerDbQueries[1]).toContain("peerdb_metric_stream_no_imu");
     expect(peerDbQueries[1]).toContain("WHERE (channel <> 'imu')");
     expect(peerDbQueries[2]).toContain("host = 'postgres.example'");
     expect(peerDbQueries[2]).toContain("port = 6543");
     expect(peerDbQueries[2]).toContain("password = 'pg''credential'");
-    expect(peerDbQueries[3]).toContain("database = 'peerdb'");
+    expect(peerDbQueries[3]).toContain("database = 'postgres_fitness'");
     expect(peerDbQueries[3]).toContain("host = 'clickhouse.example'");
     expect(peerDbQueries[3]).toContain("password = 'click\\credential'");
-    expect(peerDbQueries[4]).toContain("database = 'postgres_fitness'");
-    expect(peerDbQueries[4]).toContain("host = 'clickhouse.example'");
-    expect(peerDbQueries[4]).toContain("password = 'click\\credential'");
-    expect(peerDbQueries[5]).toContain("CREATE MIRROR IF NOT EXISTS dofek_metric_stream_cdc");
-    expect(peerDbQueries[6]).toContain("CREATE MIRROR IF NOT EXISTS dofek_metric_stream_analytics");
-    expect(peerDbQueries[6]).toContain("dofek_clickhouse_postgres_fitness");
-    expect(peerDbQueries[5]).toContain(
+    expect(peerDbQueries[4]).toContain("CREATE MIRROR IF NOT EXISTS dofek_metric_stream_analytics");
+    expect(peerDbQueries[4]).toContain("dofek_clickhouse_postgres_fitness");
+    expect(peerDbQueries[4]).toContain(
       "exclude: [device_id, source_type, vector, point, metadata]",
     );
+    expect(peerDbQueries[4]).toContain("publication_name = 'peerdb_metric_stream_no_imu'");
+    expect(peerDbQueries[5]).toContain("CREATE MIRROR IF NOT EXISTS dofek_fitness_raw_analytics");
+    expect(peerDbQueries[5]).toContain("dofek_clickhouse_postgres_fitness");
     expect(peerDbQueries[6]).toContain(
-      "exclude: [device_id, source_type, vector, point, metadata]",
-    );
-    expect(peerDbQueries[6]).toContain("publication_name = 'peerdb_metric_stream_no_imu'");
-    expect(peerDbQueries[7]).toContain("CREATE MIRROR IF NOT EXISTS dofek_fitness_raw_analytics");
-    expect(peerDbQueries[7]).toContain("dofek_clickhouse_postgres_fitness");
-    expect(peerDbQueries[8]).toContain(
       "CREATE MIRROR IF NOT EXISTS dofek_provider_inventory_raw_analytics",
     );
-    expect(peerDbQueries[8]).toContain("dofek_clickhouse_postgres_fitness");
+    expect(peerDbQueries[6]).toContain("dofek_clickhouse_postgres_fitness");
     expect(peerDbQueries.join("\n")).not.toContain("{{");
     expect(peerDbClientMocks.end).toHaveBeenCalledTimes(2);
     expect(clickHouseClientMocks.close).toHaveBeenCalledTimes(1);
@@ -569,13 +546,9 @@ describe("PeerDB ClickHouse CDC setup", () => {
       .map(([queryText]) => String(queryText))
       .filter((queryText) => !isPeerDbMirrorReconciliationQuery(queryText));
     const peerDbQueryPostgres = String(peerDbQueries[2]);
-    const peerDbQueryClickhouse = String(peerDbQueries[3]);
-    const peerDbQueryClickhousePostgresFitness = String(peerDbQueries[4]);
+    const peerDbQueryClickhousePostgresFitness = String(peerDbQueries[3]);
     expect(peerDbQueryPostgres).toContain("host = 'db'");
     expect(peerDbQueryPostgres).toContain("port = 5432");
-    expect(peerDbQueryClickhouse).toContain("host = 'clickhouse'");
-    expect(peerDbQueryClickhouse).toContain("port = 9000");
-    expect(peerDbQueryClickhouse).toContain("database = 'peerdb'");
     expect(peerDbQueryClickhousePostgresFitness).toContain("host = 'clickhouse'");
     expect(peerDbQueryClickhousePostgresFitness).toContain("port = 9000");
     expect(peerDbQueryClickhousePostgresFitness).toContain("database = 'postgres_fitness'");
@@ -608,12 +581,9 @@ describe("PeerDB ClickHouse CDC setup", () => {
       .map(([queryText]) => String(queryText))
       .filter((queryText) => !isPeerDbMirrorReconciliationQuery(queryText));
     const peerDbQueryPostgres = String(peerDbQueries[2]);
-    const peerDbQueryClickhouse = String(peerDbQueries[3]);
-    const peerDbQueryClickhousePostgresFitness = String(peerDbQueries[4]);
+    const peerDbQueryClickhousePostgresFitness = String(peerDbQueries[3]);
     expect(peerDbQueryPostgres).toContain("host = 'postgres.internal'");
     expect(peerDbQueryPostgres).toContain("port = 6545");
-    expect(peerDbQueryClickhouse).toContain("host = 'clickhouse.internal'");
-    expect(peerDbQueryClickhouse).toContain("port = 9010");
     expect(peerDbQueryClickhousePostgresFitness).toContain("host = 'clickhouse.internal'");
     expect(peerDbQueryClickhousePostgresFitness).toContain("port = 9010");
   });