Fix weekly reports and analytics worker microbatches

analytics/macros/bounded_activity_graph.sql

@@ -0,0 +1,206 @@
+{% macro bounded_activity_graph() %}
+{% set batch_start_value = config.get("__dbt_internal_microbatch_event_time_start") %}
+{% set batch_end_value = config.get("__dbt_internal_microbatch_event_time_end") %}
+{% if batch_start_value and batch_end_value %}
+{% set batch_start = batch_start_value.strftime("%Y-%m-%d %H:%M:%S") %}
+{% set batch_end = batch_end_value.strftime("%Y-%m-%d %H:%M:%S") %}
+{% endif %}
+active_activity AS (
+    SELECT *
+    FROM {{ source('postgres_fitness', 'activity') }} FINAL
+    WHERE _peerdb_is_deleted = 0
+        {% if batch_start_value and batch_end_value %}
+            AND started_at < toDateTime64('{{ batch_end }}', 6, 'UTC')
+            AND coalesce(ended_at, started_at + INTERVAL 12 HOUR) >= toDateTime64('{{ batch_start }}', 6, 'UTC')
+        {% endif %}
+),
+
+active_provider_priority AS (
+    SELECT *
+    FROM {{ source('postgres_fitness', 'provider_priority') }} FINAL
+    WHERE _peerdb_is_deleted = 0
+),
+
+active_device_priority AS (
+    SELECT *
+    FROM {{ source('postgres_fitness', 'device_priority') }} FINAL
+    WHERE _peerdb_is_deleted = 0
+),
+
+device_priority_match AS (
+    SELECT
+        activity_id,
+        priority
+    FROM (
+        SELECT
+            active_activity.id AS activity_id,
+            active_device_priority.priority AS priority,
+            row_number() OVER (
+                PARTITION BY active_activity.id
+                ORDER BY length(active_device_priority.source_name_pattern) DESC
+            ) AS row_number
+        FROM active_activity
+        INNER JOIN active_device_priority
+            ON active_device_priority.provider_id = active_activity.provider_id
+            AND active_activity.source_name LIKE active_device_priority.source_name_pattern
+    )
+    WHERE row_number = 1
+),
+
+ranked AS (
+    SELECT
+        active_activity.id AS id,
+        active_activity.provider_id AS provider_id,
+        active_activity.user_id AS user_id,
+        active_activity.external_id AS external_id,
+        active_activity.activity_type AS activity_type,
+        active_activity.started_at AS started_at,
+        active_activity.ended_at AS ended_at,
+        active_activity.source_name AS source_name,
+        active_activity.name AS name,
+        active_activity.notes AS notes,
+        active_activity.timezone AS timezone,
+        active_activity.raw AS raw,
+        coalesce(device_priority_match.priority, active_provider_priority.priority, 100) AS priority
+    FROM active_activity
+    LEFT JOIN active_provider_priority
+        ON active_provider_priority.provider_id = active_activity.provider_id
+    LEFT JOIN device_priority_match
+        ON device_priority_match.activity_id = active_activity.id
+),
+
+pairs AS (
+    SELECT
+        left_activity.id AS id1,
+        right_activity.id AS id2
+    FROM ranked AS left_activity
+    INNER JOIN ranked AS right_activity
+        ON left_activity.user_id = right_activity.user_id
+        AND toString(left_activity.id) < toString(right_activity.id)
+        AND dateDiff(
+            'second',
+            greatest(left_activity.started_at, right_activity.started_at),
+            least(
+                coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 1 HOUR),
+                coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 1 HOUR)
+            )
+        ) / nullIf(dateDiff(
+            'second',
+            least(left_activity.started_at, right_activity.started_at),
+            greatest(
+                coalesce(left_activity.ended_at, left_activity.started_at + INTERVAL 1 HOUR),
+                coalesce(right_activity.ended_at, right_activity.started_at + INTERVAL 1 HOUR)
+            )
+        ), 0) > 0.8
+),
+
+graph_edges AS (
+    SELECT
+        id1 AS from_id,
+        id2 AS to_id
+    FROM pairs
+    UNION ALL
+    SELECT
+        id2 AS from_id,
+        id1 AS to_id
+    FROM pairs
+),
+
+connected_components AS (
+    SELECT
+        id AS activity_id,
+        id AS connected_activity_id,
+        [toString(id)] AS visited_activity_ids
+    FROM ranked
+    UNION ALL
+    SELECT
+        connected_components.activity_id AS activity_id,
+        graph_edges.to_id AS connected_activity_id,
+        arrayConcat(connected_components.visited_activity_ids, [toString(graph_edges.to_id)]) AS visited_activity_ids
+    FROM connected_components
+    INNER JOIN graph_edges
+        ON graph_edges.from_id = connected_components.connected_activity_id
+    WHERE NOT has(connected_components.visited_activity_ids, toString(graph_edges.to_id))
+),
+
+final_groups AS (
+    SELECT
+        activity_id,
+        min(toString(connected_activity_id)) AS group_id
+    FROM connected_components
+    GROUP BY activity_id
+),
+
+best AS (
+    SELECT *
+    FROM (
+        SELECT
+            final_groups.group_id AS group_id,
+            ranked.id AS canonical_id,
+            ranked.provider_id AS provider_id,
+            ranked.user_id AS user_id,
+            ranked.activity_type AS activity_type,
+            ranked.started_at AS started_at,
+            ranked.ended_at AS ended_at,
+            ranked.source_name AS source_name,
+            ranked.priority AS priority,
+            row_number() OVER (
+                PARTITION BY final_groups.group_id
+                ORDER BY ranked.priority ASC, toString(ranked.id) ASC
+            ) AS row_number
+        FROM final_groups
+        INNER JOIN ranked
+            ON ranked.id = final_groups.activity_id
+    )
+    WHERE row_number = 1
+),
+
+merged AS (
+    SELECT
+        best.group_id AS group_id,
+        best.canonical_id AS id,
+        any(best.provider_id) AS provider_id,
+        any(best.user_id) AS user_id,
+        any(best.activity_type) AS activity_type,
+        any(best.started_at) AS started_at,
+        any(best.ended_at) AS ended_at,
+        any(best.source_name) AS source_name,
+        argMinIf(ranked.name, ranked.priority, ranked.name IS NOT NULL) AS name,
+        argMinIf(ranked.notes, ranked.priority, ranked.notes IS NOT NULL) AS notes,
+        argMinIf(ranked.timezone, ranked.priority, ranked.timezone IS NOT NULL) AS timezone,
+        argMinIf(ranked.raw, ranked.priority, ranked.raw IS NOT NULL) AS raw,
+        arraySort(groupUniqArray(ranked.provider_id)) AS source_providers,
+        groupArrayIf(
+            map('providerId', ranked.provider_id, 'externalId', ranked.external_id),
+            ranked.external_id IS NOT NULL AND ranked.external_id != ''
+        ) AS source_external_ids,
+        groupArray(ranked.id) AS member_activity_ids
+    FROM best
+    INNER JOIN final_groups
+        ON final_groups.group_id = best.group_id
+    INNER JOIN ranked
+        ON ranked.id = final_groups.activity_id
+    GROUP BY best.group_id, best.canonical_id
+),
+
+current_activity AS (
+    SELECT
+        id AS activity_id,
+        user_id,
+        activity_type,
+        name,
+        started_at,
+        ended_at
+    FROM merged
+),
+
+activity_members AS (
+    SELECT
+        id AS activity_id,
+        user_id,
+        started_at,
+        ended_at,
+        arrayJoin(member_activity_ids) AS member_activity_id
+    FROM merged
+)
+{% endmacro %}

analytics/models/read_models/activity_location_sample.sql

@@ -15,13 +15,7 @@
     }
 ) }}
 
-WITH activity_members AS (
-    SELECT
-        activity_id,
-        user_id,
-        member_activity_id
-    FROM {{ source('analytics', 'v_activity_members') }}
-),
+WITH RECURSIVE {{ bounded_activity_graph() }},
 
 location_versions AS (
     SELECT *

analytics/models/read_models/activity_sensor_sample.sql

@@ -15,14 +15,7 @@
     }
 ) }}
 
-WITH current_activity AS (
-    SELECT
-        id AS activity_id,
-        user_id,
-        started_at,
-        coalesce(ended_at, started_at + INTERVAL 12 HOUR) AS ended_at
-    FROM {{ source('analytics', 'v_activity') }}
-),
+WITH RECURSIVE {{ bounded_activity_graph() }},
 
 activity_samples AS (
     SELECT
@@ -37,7 +30,7 @@ activity_samples AS (
     INNER JOIN current_activity
         ON current_activity.user_id = samples.user_id
         AND samples.recorded_at >= current_activity.started_at
-        AND samples.recorded_at <= current_activity.ended_at
+        AND samples.recorded_at <= coalesce(current_activity.ended_at, current_activity.started_at + INTERVAL 12 HOUR)
 )
 
 SELECT

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

@@ -14,6 +14,14 @@ function compactWhitespace(value: string): string {
 }
 
 describe("production analytics read-model build", () => {
+  it("does not block the BullMQ worker on analytics dbt builds", () => {
+    const entrypoint = readProjectFile("entrypoint.sh");
+    const workerBlockMatch = entrypoint.match(/  worker\)\n(?<body>[\s\S]*?)\n    ;;/);
+
+    expect(workerBlockMatch?.groups?.body).toContain("exec $NODE src/jobs/worker.ts");
+    expect(workerBlockMatch?.groups?.body).not.toContain("dbt build");
+  });
+
   it("runs every bounded intermediary and final read model in dependency order", () => {
     const entrypoint = readProjectFile("entrypoint.sh");
     const safeModelMatch = entrypoint.match(/^DBT_SAFE_MODELS="([^"]+)"$/m);
@@ -60,7 +68,9 @@ describe("production analytics read-model build", () => {
     expect(sql).toContain("incremental_strategy='microbatch'");
     expect(sql).toContain("event_time='recorded_at'");
     expect(sql).toContain("ref('deduped_sensor')");
-    expect(sql).toContain("source('analytics', 'v_activity')");
+    expect(sql).toContain("WITH RECURSIVE {{ bounded_activity_graph() }}");
+    expect(sql).toContain("bounded_activity_graph()");
+    expect(sql).not.toContain("source('analytics', 'v_activity')");
     expect(sql).toContain("activity_id");
   });
 
@@ -71,10 +81,23 @@ describe("production analytics read-model build", () => {
     expect(sql).toContain("incremental_strategy='microbatch'");
     expect(sql).toContain("event_time='recorded_at'");
     expect(sql).toContain("source('postgres_fitness', 'metric_stream')");
-    expect(sql).toContain("source('analytics', 'v_activity_members')");
+    expect(sql).toContain("WITH RECURSIVE {{ bounded_activity_graph() }}");
+    expect(sql).toContain("bounded_activity_graph()");
+    expect(sql).not.toContain("source('analytics', 'v_activity_members')");
     expect(sql).toContain("channel = 'location'");
   });
 
+  it("bounds activity graph construction to the active microbatch window", () => {
+    const sql = readProjectFile("analytics/macros/bounded_activity_graph.sql");
+
+    expect(sql).toContain("__dbt_internal_microbatch_event_time_start");
+    expect(sql).toContain("__dbt_internal_microbatch_event_time_end");
+    expect(sql).toContain("started_at < toDateTime64('{{ batch_end }}', 6, 'UTC')");
+    expect(sql).toContain(
+      "coalesce(ended_at, started_at + INTERVAL 12 HOUR) >= toDateTime64('{{ batch_start }}', 6, 'UTC')",
+    );
+  });
+
   it("aggregates activity sensor summary from the bounded sensor intermediary", () => {
     expect(existsSync(new URL("./activity_sensor_summary_rows.sql", import.meta.url))).toBe(true);
     const sql = readModel("activity_sensor_summary_rows");

analytics/models/sources.yml

@@ -10,9 +10,11 @@ sources:
     schema: postgres_fitness
     tables:
       - name: activity
+      - name: device_priority
       - name: metric_stream
         config:
           event_time: recorded_at
+      - name: provider_priority
       - name: sensor_device_priority
       - name: sensor_provider_priority
       - name: sleep_session

docs/production-incident-baseline.md

@@ -8796,3 +8796,44 @@ new incremental tables are populated.
 - Follow-Up Work: Deploy the prepared bounded RHR and activity summary model
   update; add freshness monitoring comparing latest sleep/heart-rate inputs to
   latest active RHR output date.
+
+### Production worker restart loop from analytics dbt OOM
+
+- Date: 2026-05-26 PT / 2026-05-27 UTC.
+- Symptoms: Production was reported down. Public `/healthz` and the SPA shell
+  still returned HTTP 200, but background processing was impaired: the
+  `dofek_worker` service was repeatedly failing with `task: non-zero exit (1)`,
+  and `dofek_analytics-worker` was retrying dbt builds.
+- User Impact: Background sync and import jobs were unavailable while the
+  worker was in a restart loop. ClickHouse-backed dashboard data was at risk of
+  stale or failed reads while ClickHouse was under memory pressure.
+- Evidence: The first fatal analytics-worker log lines were ClickHouse
+  `MEMORY_LIMIT_EXCEEDED` exceptions in dbt models including
+  `activity_location_sample` and `activity_sensor_sample`, with ClickHouse at
+  its configured 3 GiB memory limit while executing recursive activity graph
+  work. The failing batch for `2026-05-20` had only 458 heart-rate sample rows
+  but used `analytics.v_activity`/`analytics.v_activity_members`, which forced
+  ClickHouse to traverse the unbounded 1,029-activity user graph before the
+  microbatch could finish. The worker logs also showed the `worker` entrypoint
+  running Postgres migrations, ClickHouse migrations, and then `dbt build`
+  before the BullMQ worker process could start.
+- Root Cause: The activity sample dbt models used global recursive activity
+  views inside each microbatch. ClickHouse did not push the day filter inside
+  those views, so each batch recomputed the full dedupe graph and OOMed. The
+  blast radius was larger because `entrypoint.sh` also coupled the BullMQ
+  `worker` service startup to the same analytics dbt build.
+- Fix/Mitigation: A short-lived manual mitigation stopped the failing dbt retry
+  loop and temporarily ran the Node BullMQ worker directly so background jobs
+  could resume. The durable code change removes `dbt build` from the `worker`
+  entrypoint and bounds the activity graph inside the affected analytics models
+  to the active microbatch window so `analytics-worker` remains enabled.
+- Validation: After the current production image was redeployed,
+  `dofek_web` was `2/2`, `dofek_worker` was `1/1`, and
+  `dofek_analytics-worker` was `1/1`. `/healthz` returned HTTP 200. The local
+  fix compiles the affected dbt models with batch-local activity graph bounds
+  and passes the changed test suite.
+- Remaining Risk: Medium until the bounded activity graph fix is deployed and
+  the next scheduled analytics-worker run completes cleanly on production-scale
+  data.
+- Follow-Up Work: Add freshness monitoring for analytics read models and keep
+  the BullMQ worker independent from analytics rebuilds.

entrypoint.sh

@@ -31,7 +31,6 @@ case "${1:-sync}" in
     ;;
   worker)
     $NODE src/db/run-migrate.ts
-    dbt build --project-dir analytics --profiles-dir analytics --threads 1 --select $DBT_SAFE_MODELS
     exec $NODE src/jobs/worker.ts
     ;;
   migrate)