Expire recovery scores after source deletion

analytics/models/read_models/daily_recovery.sql

@@ -4,44 +4,25 @@
     engine='ReplacingMergeTree(refresh_version)',
     order_by='(user_id, date)',
     query_settings={
-        'max_threads': 1
+        'max_threads': 1,
+        'join_use_nulls': 1
     }
 ) }}
 
 WITH {% if is_incremental() %}
-existing_dates AS (
-    SELECT
-        user_id,
-        max(date) AS latest_materialized_date,
-        max(refreshed_at) AS latest_materialized_refreshed_at
+target_state AS (
+    SELECT coalesce(max(refreshed_at), toDateTime64(0, 9, 'UTC')) AS last_refreshed_at
     FROM {{ this }}
-    GROUP BY user_id
 ),
-{% endif %}
 
-recovery_inputs AS (
-    SELECT
-        user_id,
-        date,
-        hrv,
-        resting_hr,
-        respiratory_rate,
-        efficiency_pct,
-        hrv_mean_30d,
-        hrv_sd_30d,
-        rhr_mean_30d,
-        rhr_sd_30d,
-        rr_mean_30d,
-        rr_sd_30d,
-        hrv_mean_60d,
-        hrv_sd_60d,
-        rhr_mean_60d,
-        rhr_sd_60d,
-        refreshed_at
-    FROM {{ ref('daily_recovery_inputs') }} FINAL
+changed_users AS (
+    SELECT DISTINCT recovery_inputs.user_id AS user_id
+    FROM {{ ref('daily_recovery_inputs') }} AS recovery_inputs FINAL
+    WHERE recovery_inputs.refreshed_at > (SELECT last_refreshed_at FROM target_state)
 ),
+{% endif %}
 
-recovery_inputs_to_materialize AS (
+recovery_inputs AS (
     SELECT
         recovery_inputs.user_id AS user_id,
         recovery_inputs.date AS date,
@@ -59,14 +40,11 @@ recovery_inputs_to_materialize AS (
         recovery_inputs.hrv_sd_60d AS hrv_sd_60d,
         recovery_inputs.rhr_mean_60d AS rhr_mean_60d,
         recovery_inputs.rhr_sd_60d AS rhr_sd_60d
-    FROM recovery_inputs
-    {% if is_incremental() %}
-    LEFT JOIN existing_dates
-        ON existing_dates.user_id = recovery_inputs.user_id
-    WHERE existing_dates.user_id IS NULL
-        OR recovery_inputs.refreshed_at > existing_dates.latest_materialized_refreshed_at
-        OR recovery_inputs.date >= existing_dates.latest_materialized_date - INTERVAL 60 DAY
-    {% endif %}
+    FROM {{ ref('daily_recovery_inputs') }} AS recovery_inputs FINAL
+    WHERE recovery_inputs.is_deleted = 0
+        {% if is_incremental() %}
+        AND user_id IN (SELECT user_id FROM changed_users)
+        {% endif %}
 ),
 
 scored AS (
@@ -92,7 +70,7 @@ scored AS (
             least(100, greatest(0, round(efficiency_pct))),
             62
         ) AS sleep_score
-    FROM recovery_inputs_to_materialize
+    FROM recovery_inputs
 ),
 
 sigmoid_inputs AS (
@@ -137,34 +115,63 @@ sigmoid_scores AS (
     FROM sigmoid_inputs
 ),
 
+{% if is_incremental() %}
+existing_keys AS (
+    SELECT
+        user_id,
+        date
+    FROM {{ this }} FINAL
+    WHERE is_deleted = 0
+        AND user_id IN (SELECT user_id FROM changed_users)
+),
+{% endif %}
+
+result_keys AS (
+    SELECT
+        user_id,
+        date
+    FROM sigmoid_scores
+    {% if is_incremental() %}
+    UNION DISTINCT
+    SELECT
+        user_id,
+        date
+    FROM existing_keys
+    {% endif %}
+),
+
 refresh_clock AS (
     SELECT
         toUInt64(toUnixTimestamp64Nano(now64(9))) AS refresh_version,
-        now64(9) AS refreshed_at
+        now64(9, 'UTC') AS refreshed_at
 )
 
 SELECT
-    CAST(user_id, 'UUID') AS user_id,
-    CAST(date, 'Date') AS date,
-    hrv,
-    resting_hr,
-    respiratory_rate,
-    efficiency_pct,
-    hrv_mean_30d,
-    hrv_sd_30d,
-    rhr_mean_30d,
-    rhr_sd_30d,
-    rr_mean_30d,
-    rr_sd_30d,
-    hrv_mean_60d,
-    hrv_sd_60d,
-    rhr_mean_60d,
-    rhr_sd_60d,
-    hrv_score,
-    resting_hr_score,
-    sleep_score,
-    respiratory_rate_score,
+    CAST(result_keys.user_id, 'UUID') AS user_id,
+    CAST(result_keys.date, 'Date') AS date,
+    sigmoid_scores.hrv AS hrv,
+    sigmoid_scores.resting_hr AS resting_hr,
+    sigmoid_scores.respiratory_rate AS respiratory_rate,
+    sigmoid_scores.efficiency_pct AS efficiency_pct,
+    sigmoid_scores.hrv_mean_30d AS hrv_mean_30d,
+    sigmoid_scores.hrv_sd_30d AS hrv_sd_30d,
+    sigmoid_scores.rhr_mean_30d AS rhr_mean_30d,
+    sigmoid_scores.rhr_sd_30d AS rhr_sd_30d,
+    sigmoid_scores.rr_mean_30d AS rr_mean_30d,
+    sigmoid_scores.rr_sd_30d AS rr_sd_30d,
+    sigmoid_scores.hrv_mean_60d AS hrv_mean_60d,
+    sigmoid_scores.hrv_sd_60d AS hrv_sd_60d,
+    sigmoid_scores.rhr_mean_60d AS rhr_mean_60d,
+    sigmoid_scores.rhr_sd_60d AS rhr_sd_60d,
+    sigmoid_scores.hrv_score AS hrv_score,
+    sigmoid_scores.resting_hr_score AS resting_hr_score,
+    sigmoid_scores.sleep_score AS sleep_score,
+    sigmoid_scores.respiratory_rate_score AS respiratory_rate_score,
+    if(sigmoid_scores.user_id IS NULL, 1, 0) AS is_deleted,
     refresh_clock.refresh_version AS refresh_version,
     refresh_clock.refreshed_at AS refreshed_at
-FROM sigmoid_scores
+FROM result_keys
+LEFT JOIN sigmoid_scores
+    ON sigmoid_scores.user_id = result_keys.user_id
+    AND sigmoid_scores.date = result_keys.date
 CROSS JOIN refresh_clock

analytics/models/read_models/daily_recovery_inputs.sql

@@ -4,11 +4,36 @@
     engine='ReplacingMergeTree(refresh_version)',
     order_by='(user_id, date)',
     query_settings={
-        'max_threads': 1
+        'max_threads': 1,
+        'join_use_nulls': 1
     }
 ) }}
 
-WITH daily_metrics AS (
+WITH {% if is_incremental() %}
+existing_rows AS (
+    SELECT
+        user_id,
+        date,
+        hrv,
+        resting_hr,
+        respiratory_rate,
+        efficiency_pct,
+        hrv_mean_30d,
+        hrv_sd_30d,
+        rhr_mean_30d,
+        rhr_sd_30d,
+        rr_mean_30d,
+        rr_sd_30d,
+        hrv_mean_60d,
+        hrv_sd_60d,
+        rhr_mean_60d,
+        rhr_sd_60d
+    FROM {{ this }} FINAL
+    WHERE is_deleted = 0
+),
+{% endif %}
+
+daily_metrics AS (
     SELECT
         user_id,
         date,
@@ -117,15 +142,82 @@ inputs_with_baselines AS (
     FROM daily_inputs
 ),
 
+{% if is_incremental() %}
+dirty_keys AS (
+    SELECT
+        inputs_with_baselines.user_id AS user_id,
+        inputs_with_baselines.date AS date
+    FROM inputs_with_baselines
+    LEFT JOIN existing_rows
+        ON existing_rows.user_id = inputs_with_baselines.user_id
+        AND existing_rows.date = inputs_with_baselines.date
+    WHERE existing_rows.user_id IS NULL
+        OR tuple(
+            inputs_with_baselines.hrv,
+            inputs_with_baselines.resting_hr,
+            inputs_with_baselines.respiratory_rate,
+            inputs_with_baselines.efficiency_pct,
+            inputs_with_baselines.hrv_mean_30d,
+            inputs_with_baselines.hrv_sd_30d,
+            inputs_with_baselines.rhr_mean_30d,
+            inputs_with_baselines.rhr_sd_30d,
+            inputs_with_baselines.rr_mean_30d,
+            inputs_with_baselines.rr_sd_30d,
+            inputs_with_baselines.hrv_mean_60d,
+            inputs_with_baselines.hrv_sd_60d,
+            inputs_with_baselines.rhr_mean_60d,
+            inputs_with_baselines.rhr_sd_60d
+        ) IS DISTINCT FROM tuple(
+            existing_rows.hrv,
+            existing_rows.resting_hr,
+            existing_rows.respiratory_rate,
+            existing_rows.efficiency_pct,
+            existing_rows.hrv_mean_30d,
+            existing_rows.hrv_sd_30d,
+            existing_rows.rhr_mean_30d,
+            existing_rows.rhr_sd_30d,
+            existing_rows.rr_mean_30d,
+            existing_rows.rr_sd_30d,
+            existing_rows.hrv_mean_60d,
+            existing_rows.hrv_sd_60d,
+            existing_rows.rhr_mean_60d,
+            existing_rows.rhr_sd_60d
+        )
+    UNION DISTINCT
+    SELECT
+        existing_rows.user_id AS user_id,
+        existing_rows.date AS date
+    FROM existing_rows
+    LEFT JOIN inputs_with_baselines
+        ON inputs_with_baselines.user_id = existing_rows.user_id
+        AND inputs_with_baselines.date = existing_rows.date
+    WHERE inputs_with_baselines.user_id IS NULL
+),
+{% endif %}
+
+result_keys AS (
+    {% if is_incremental() %}
+    SELECT
+        user_id,
+        date
+    FROM dirty_keys
+    {% else %}
+    SELECT
+        user_id,
+        date
+    FROM inputs_with_baselines
+    {% endif %}
+),
+
 refresh_clock AS (
     SELECT
         toUInt64(toUnixTimestamp64Nano(now64(9))) AS refresh_version,
-        now64(9) AS refreshed_at
+        now64(9, 'UTC') AS refreshed_at
 )
 
 SELECT
-    CAST(inputs_with_baselines.user_id, 'UUID') AS user_id,
-    CAST(inputs_with_baselines.date, 'Date') AS date,
+    CAST(result_keys.user_id, 'UUID') AS user_id,
+    CAST(result_keys.date, 'Date') AS date,
     inputs_with_baselines.hrv AS hrv,
     inputs_with_baselines.resting_hr AS resting_hr,
     inputs_with_baselines.respiratory_rate AS respiratory_rate,
@@ -140,7 +232,11 @@ SELECT
     inputs_with_baselines.hrv_sd_60d AS hrv_sd_60d,
     inputs_with_baselines.rhr_mean_60d AS rhr_mean_60d,
     inputs_with_baselines.rhr_sd_60d AS rhr_sd_60d,
+    if(inputs_with_baselines.user_id IS NULL, 1, 0) AS is_deleted,
     refresh_clock.refresh_version AS refresh_version,
     refresh_clock.refreshed_at AS refreshed_at
-FROM inputs_with_baselines
+FROM result_keys
+LEFT JOIN inputs_with_baselines
+    ON inputs_with_baselines.user_id = result_keys.user_id
+    AND inputs_with_baselines.date = result_keys.date
 CROSS JOIN refresh_clock

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

@@ -728,12 +728,17 @@ describe("production analytics read-model build", () => {
   it("materializes daily recovery inputs from compact daily and sleep sources", () => {
     const sql = readModel("daily_recovery_inputs");
 
+    expect(sql).toContain("{% if is_incremental() %}");
+    expect(sql).toContain("existing_rows AS");
+    expect(sql).toContain("dirty_keys AS");
+    expect(sql).toContain("IS DISTINCT FROM tuple(");
     expect(sql).toContain("analytics.v_daily_metrics");
     expect(sql).toContain("analytics.v_sleep");
     expect(sql).toContain("argMax(efficiency_pct, tuple(duration_minutes, started_at))");
     expect(sql).toContain("ref('resting_heart_rate_sleep_window')");
     expect(sql).toContain("hrv_mean_60d");
     expect(sql).toContain("rhr_mean_60d");
+    expect(sql).toContain("if(inputs_with_baselines.user_id IS NULL, 1, 0) AS is_deleted");
     expect(sql).not.toContain("source('postgres_fitness', 'metric_stream')");
     expect(sql).not.toContain("ref('deduped_sensor')");
   });
@@ -745,10 +750,13 @@ describe("production analytics read-model build", () => {
     expect(sql).toContain("materialized='incremental'");
     expect(sql).toContain("engine='ReplacingMergeTree(refresh_version)'");
     expect(sql).toContain("{% if is_incremental() %}");
-    expect(sql).toContain("existing_dates AS");
-    expect(sql).toContain("latest_materialized_refreshed_at");
-    expect(normalizedSql).toContain("recovery_inputs.refreshed_at > existing_dates.latest_materialized_refreshed_at");
-    expect(normalizedSql).toContain("existing_dates.latest_materialized_date - INTERVAL 60 DAY");
+    expect(sql).toContain("changed_users AS");
+    expect(sql).toContain("existing_keys AS");
+    expect(normalizedSql).toContain("WHERE recovery_inputs.is_deleted = 0");
+    expect(normalizedSql).toContain(
+      "recovery_inputs.refreshed_at > (SELECT last_refreshed_at FROM target_state)",
+    );
+    expect(sql).toContain("if(sigmoid_scores.user_id IS NULL, 1, 0) AS is_deleted");
     expect(sql).toContain("ref('daily_recovery_inputs')");
     expect(sql).toContain("hrv_score");
     expect(sql).toContain("resting_hr_score");

packages/server/src/repositories/stress-repository.test.ts

@@ -51,6 +51,7 @@ describe("StressRepository", () => {
 
       const queryText = sensorStore.query.mock.calls[0]?.[1];
       expect(queryText).toContain("analytics.daily_recovery AS recovery_inputs FINAL");
+      expect(queryText).toContain("recovery_inputs.is_deleted = 0");
       expect(queryText).not.toContain("fitness.v_daily_metrics");
       expect(queryText).not.toContain("analytics.v_sleep");
     });

packages/server/src/repositories/stress-repository.ts

@@ -133,6 +133,7 @@ export class StressRepository extends BaseRepository {
         efficiency_pct
       FROM analytics.daily_recovery AS recovery_inputs FINAL
       WHERE recovery_inputs.user_id = {userId:UUID}
+        AND recovery_inputs.is_deleted = 0
         ${clickHouseWindowStartPredicate({
           expression: "recovery_inputs.date",
           days,

packages/server/src/routers/clickhouse-integration-test-models.ts

@@ -150,6 +150,7 @@ hrv_mean_60d Nullable(Float64),
 hrv_sd_60d Nullable(Float64),
 rhr_mean_60d Nullable(Float64),
 rhr_sd_60d Nullable(Float64),
+is_deleted UInt8 DEFAULT 0,
 refresh_version UInt64,
 refreshed_at DateTime64(9)`,
     daily_recovery: `user_id UUID,
@@ -172,6 +173,7 @@ hrv_score Nullable(Float64),
 resting_hr_score Nullable(Float64),
 sleep_score Nullable(Float64),
 respiratory_rate_score Nullable(Float64),
+is_deleted UInt8 DEFAULT 0,
 refresh_version UInt64,
 refreshed_at DateTime64(9)`,
     daily_body_measurement: `measurement_id UUID,

packages/server/src/routers/clickhouse-integration-test-read-models-a.ts

@@ -689,6 +689,7 @@ SELECT
   inputs_with_baselines.hrv_sd_60d AS hrv_sd_60d,
   inputs_with_baselines.rhr_mean_60d AS rhr_mean_60d,
   inputs_with_baselines.rhr_sd_60d AS rhr_sd_60d,
+  0 AS is_deleted,
   refresh_clock.refresh_version AS refresh_version,
   refresh_clock.refreshed_at AS refreshed_at
 FROM inputs_with_baselines

packages/server/src/routers/clickhouse-integration-test-read-models-b.ts

@@ -396,9 +396,11 @@ export function buildTestRecoveryReadModelSelectSql(
   CAST(NULL, 'Nullable(Float64)') AS resting_hr_score,
   CAST(NULL, 'Nullable(Float64)') AS sleep_score,
   CAST(NULL, 'Nullable(Float64)') AS respiratory_rate_score,
+  0 AS is_deleted,
   refresh_version,
   refreshed_at
-FROM ${databases.analytics}.daily_recovery_inputs`;
+FROM ${databases.analytics}.daily_recovery_inputs
+WHERE is_deleted = 0`;
 }
 
 export function buildTestStrainReadModelSelectSql(databases: IsolatedClickHouseDatabases): string {