Add self-healing for stale activity materialized views

packages/mobile/app/activities.tsx

@@ -84,6 +84,8 @@ export default function ActivitiesScreen() {
         ListEmptyComponent={
           query.isLoading ? (
             <ActivityIndicator color={colors.accent} style={styles.loader} />
+          ) : query.isError ? (
+            <Text style={styles.error}>Failed to load activities.</Text>
           ) : (
             <Text style={styles.empty}>No activities found</Text>
           )
@@ -156,6 +158,12 @@ const styles = StyleSheet.create({
     marginTop: 40,
     fontSize: 14,
   },
+  error: {
+    color: "#f87171",
+    textAlign: "center",
+    marginTop: 40,
+    fontSize: 14,
+  },
   pagination: {
     flexDirection: "row",
     alignItems: "center",

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

@@ -335,6 +335,28 @@ export class ActivityRepository extends BaseRepository {
     return mapHrZones(rows);
   }
 
+  /** Count activities in the base table (not the materialized view) for this user within a time window. */
+  async baseTableCount(endDate: string, days: number): Promise<number> {
+    const rows = await this.query(
+      z.object({ count: z.coerce.number() }),
+      sql`SELECT count(*)::int AS count FROM fitness.activity
+          WHERE user_id = ${this.userId}
+            AND started_at > ${timestampWindowStart(endDate, days)}`,
+    );
+    return rows[0]?.count ?? 0;
+  }
+
+  /** Refresh the activity-related materialized views. */
+  async refreshActivityViews(): Promise<void> {
+    for (const view of ["fitness.v_activity", "fitness.activity_summary"]) {
+      try {
+        await this.db.execute(sql.raw(`REFRESH MATERIALIZED VIEW CONCURRENTLY ${view}`));
+      } catch {
+        await this.db.execute(sql.raw(`REFRESH MATERIALIZED VIEW ${view}`));
+      }
+    }
+  }
+
   /** Delete an activity by ID. */
   async delete(activityId: string): Promise<void> {
     await this.db.execute(sql`

packages/server/src/routers/activity.test.ts

@@ -33,6 +33,15 @@ vi.mock("../lib/typed-sql.ts", async (importOriginal) => {
   };
 });
 
+vi.mock("@sentry/node", () => ({
+  captureMessage: vi.fn(),
+  captureException: vi.fn(),
+}));
+
+vi.mock("../logger.ts", () => ({
+  logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() },
+}));
+
 vi.mock("./sync.ts", () => ({
   ensureProvidersRegistered: vi.fn(async () => {}),
 }));
@@ -138,9 +147,77 @@ describe("activityRouter", () => {
     });
 
     it("returns empty items and zero totalCount when no activities", async () => {
-      const caller = makeCaller([]);
+      const execute = vi.fn().mockResolvedValue([]);
+      const caller = createCaller({
+        db: { execute },
+        userId: "user-1",
+        timezone: "UTC",
+      });
+      const result = await caller.list({ days: 30 });
+      expect(result).toEqual({ items: [], totalCount: 0 });
+      // Should check base table when view returns empty
+      expect(execute).toHaveBeenCalledTimes(2); // list query + base table check
+    });
+
+    it("refreshes stale views and retries when view is empty but base table has data", async () => {
+      const activityRow = {
+        id: "a1",
+        started_at: "2024-01-01 10:00:00+00",
+        ended_at: "2024-01-01 11:00:00+00",
+        activity_type: "cycling",
+        name: "Morning Ride",
+        provider_id: "wahoo",
+        source_providers: ["wahoo"],
+        avg_hr: 150,
+        max_hr: 180,
+        avg_power: 200,
+        distance_meters: 30000,
+        total_count: 1,
+      };
+      const execute = vi
+        .fn()
+        .mockResolvedValueOnce([]) // 1. list from v_activity: empty
+        .mockResolvedValueOnce([{ count: 1 }]) // 2. base table count: has data
+        .mockResolvedValueOnce([]) // 3. REFRESH v_activity
+        .mockResolvedValueOnce([]) // 4. REFRESH activity_summary
+        .mockResolvedValueOnce([activityRow]); // 5. retry list
+      const caller = createCaller({
+        db: { execute },
+        userId: "user-1",
+        timezone: "UTC",
+      });
+      const result = await caller.list({ days: 30, limit: 20, offset: 0 });
+      expect(result.items).toHaveLength(1);
+      expect(result.items[0]).toMatchObject({ id: "a1" });
+      expect(execute).toHaveBeenCalledTimes(5);
+    });
+
+    it("returns empty when both view and base table are empty (genuinely no data)", async () => {
+      const execute = vi
+        .fn()
+        .mockResolvedValueOnce([]) // 1. list from v_activity: empty
+        .mockResolvedValueOnce([{ count: 0 }]); // 2. base table count: no data
+      const caller = createCaller({
+        db: { execute },
+        userId: "user-1",
+        timezone: "UTC",
+      });
       const result = await caller.list({ days: 30 });
       expect(result).toEqual({ items: [], totalCount: 0 });
+      expect(execute).toHaveBeenCalledTimes(2); // no refresh or retry
+    });
+
+    it("skips stale view check on non-first pages", async () => {
+      const execute = vi.fn().mockResolvedValue([]);
+      const caller = createCaller({
+        db: { execute },
+        userId: "user-1",
+        timezone: "UTC",
+      });
+      const result = await caller.list({ days: 30, limit: 20, offset: 20 });
+      expect(result).toEqual({ items: [], totalCount: 0 });
+      // Only the list query — no base table check on offset > 0
+      expect(execute).toHaveBeenCalledTimes(1);
     });
 
     it("uses default limit of 20 and offset of 0", async () => {
@@ -151,8 +228,8 @@ describe("activityRouter", () => {
         timezone: "UTC",
       });
       await caller.list({ days: 30 });
-      // Verify the query was called (default params applied)
-      expect(execute).toHaveBeenCalledTimes(1);
+      // list query + base table count (stale view check)
+      expect(execute).toHaveBeenCalledTimes(2);
     });
   });
 

packages/server/src/routers/activity.ts

@@ -1,7 +1,9 @@
+import * as Sentry from "@sentry/node";
 import { TRPCError } from "@trpc/server";
 import { getProvider } from "dofek/providers/registry";
 import { z } from "zod";
 import { endDateSchema } from "../lib/date-window.ts";
+import { logger } from "../logger.ts";
 import { Activity, type ActivityDetail } from "../models/activity.ts";
 import {
   ActivityRepository,
@@ -45,7 +47,42 @@ export const activityRouter = router({
     )
     .query(async ({ ctx, input }) => {
       const repo = new ActivityRepository(ctx.db, ctx.userId, ctx.timezone);
-      return repo.list(input);
+      const result = await repo.list(input);
+
+      // Self-healing: if the materialized view returns no results on the first
+      // page but the base table has data in the same time window, the views are
+      // stale (e.g. after a crash recovery or failed view refresh). Refresh
+      // them and retry the query. Only check on the first page to avoid
+      // expensive refreshes on legitimate empty later pages.
+      if (input.offset === 0 && result.items.length === 0) {
+        const baseCount = await repo.baseTableCount(input.endDate, input.days);
+        if (baseCount > 0) {
+          logger.warn(
+            `[activity] Stale views detected for user ${ctx.userId}: ` +
+              `${baseCount} activities in base table but 0 in materialized view. Refreshing.`,
+          );
+          Sentry.captureMessage("Stale activity materialized views detected", {
+            level: "warning",
+            tags: { userId: ctx.userId },
+            extra: { baseCount },
+          });
+          try {
+            await repo.refreshActivityViews();
+            return repo.list(input);
+          } catch (refreshError) {
+            const errorDetail =
+              refreshError instanceof Error
+                ? (refreshError.stack ?? refreshError.message)
+                : String(refreshError);
+            logger.error(`[activity] Failed to refresh stale views: ${errorDetail}`);
+            Sentry.captureException(refreshError, {
+              tags: { userId: ctx.userId, context: "staleViewRefresh" },
+            });
+          }
+        }
+      }
+
+      return result;
     }),
 
   byId: cachedProtectedQuery(CacheTTL.MEDIUM)

packages/server/src/routers/admin.test.ts

@@ -12,6 +12,10 @@ vi.mock("../lib/start-worker.ts", () => ({
   startWorker: vi.fn(),
 }));
 
+vi.mock("../logger.ts", () => ({
+  logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() },
+}));
+
 vi.mock("../trpc.ts", async () => {
   const { initTRPC } = await import("@trpc/server");
   const trpc = initTRPC
@@ -436,6 +440,36 @@ describe("adminRouter", () => {
     });
   });
 
+  describe("refreshViews", () => {
+    it("refreshes all materialized views and returns view names", async () => {
+      const execute = vi.fn().mockResolvedValue([]);
+      const caller = makeCaller(execute);
+      const result = await caller.refreshViews();
+      expect(result.refreshed).toEqual([
+        "fitness.v_activity",
+        "fitness.v_sleep",
+        "fitness.v_body_measurement",
+        "fitness.v_daily_metrics",
+        "fitness.activity_summary",
+      ]);
+      // 5 views × REFRESH MATERIALIZED VIEW CONCURRENTLY
+      expect(execute).toHaveBeenCalledTimes(5);
+    });
+
+    it("falls back to non-concurrent refresh on error", async () => {
+      const execute = vi
+        .fn()
+        .mockRejectedValueOnce(new Error("has not been populated"))
+        .mockResolvedValueOnce([]) // fallback non-concurrent
+        .mockResolvedValue([]); // remaining views
+      const caller = makeCaller(execute);
+      const result = await caller.refreshViews();
+      expect(result.refreshed).toHaveLength(5);
+      // 1 failed concurrent + 1 fallback + 4 remaining = 6
+      expect(execute).toHaveBeenCalledTimes(6);
+    });
+  });
+
   describe("trainingExportStatus", () => {
     it("returns watermark data", async () => {
       const rows = [

packages/server/src/routers/admin.ts

@@ -3,8 +3,17 @@ import { sql } from "drizzle-orm";
 import { z } from "zod";
 import { startWorker } from "../lib/start-worker.ts";
 import { executeWithSchema, timestampStringSchema } from "../lib/typed-sql.ts";
+import { logger } from "../logger.ts";
 import { adminProcedure, router } from "../trpc.ts";
 
+const ALL_MATERIALIZED_VIEWS = [
+  "fitness.v_activity",
+  "fitness.v_sleep",
+  "fitness.v_body_measurement",
+  "fitness.v_daily_metrics",
+  "fitness.activity_summary",
+] as const;
+
 const trainingExportQueue = createTrainingExportQueue();
 
 // ── Schemas for admin queries ──
@@ -443,6 +452,22 @@ export const adminRouter = router({
       return { jobId: String(job.id) };
     }),
 
+  /** Force-refresh all materialized views (dedup + rollup). */
+  refreshViews: adminProcedure.mutation(async ({ ctx }) => {
+    logger.info("[admin] Refreshing all materialized views");
+    const refreshed: string[] = [];
+    for (const view of ALL_MATERIALIZED_VIEWS) {
+      try {
+        await ctx.db.execute(sql.raw(`REFRESH MATERIALIZED VIEW CONCURRENTLY ${view}`));
+      } catch {
+        await ctx.db.execute(sql.raw(`REFRESH MATERIALIZED VIEW ${view}`));
+      }
+      refreshed.push(view);
+    }
+    logger.info(`[admin] Refreshed ${refreshed.length} materialized views`);
+    return { refreshed };
+  }),
+
   /** Get training export watermark status */
   trainingExportStatus: adminProcedure.query(async ({ ctx }) => {
     const watermarkSchema = z.object({

packages/web/src/components/ActivityList.test.tsx

@@ -65,6 +65,12 @@ describe("ActivityList", () => {
     expect(screen.getByText("No recent activities")).toBeDefined();
   });
 
+  it("shows error state when error prop is true", () => {
+    renderWithUnits(<ActivityList activities={[]} error={true} />);
+    expect(screen.getByText("Failed to load activities.")).toBeDefined();
+    expect(screen.queryByText("No recent activities")).toBeNull();
+  });
+
   it("renders loading state", () => {
     renderWithUnits(<ActivityList activities={[]} loading={true} />);
     // ChartLoadingSkeleton should be visible

packages/web/src/components/ActivityList.tsx

@@ -19,6 +19,7 @@ export interface Activity {
 interface ActivityListProps {
   activities: Activity[];
   loading?: boolean;
+  error?: boolean;
   totalCount?: number;
   page?: number;
   pageSize?: number;
@@ -42,6 +43,7 @@ function formatActivityDuration(startedAt: string, endedAt: string | null): stri
 export function ActivityList({
   activities,
   loading,
+  error,
   totalCount,
   page,
   pageSize,
@@ -54,6 +56,10 @@ export function ActivityList({
     return <ChartLoadingSkeleton height={100} />;
   }
 
+  if (error) {
+    return <p className="text-sm text-red-400 py-4">Failed to load activities.</p>;
+  }
+
   if (activities.length === 0) {
     return <div className="text-subtle text-sm py-4">No recent activities</div>;
   }

packages/web/src/pages/Dashboard.tsx

@@ -475,6 +475,7 @@ export function Dashboard() {
           <ActivityList
             activities={assertRows(activities.data?.items, activityRowSchema)}
             loading={activities.isLoading}
+            error={activities.isError}
             totalCount={activities.data?.totalCount}
             page={activityPage}
             pageSize={activityPageSize}