Report BullMQ progress for all jobs

src/jobs/process-activity-delete-analytics-job.test.ts

@@ -1,11 +1,20 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
+const mockCaptureException = vi.fn();
+vi.mock("@sentry/node", () => ({
+  captureException: (...args: unknown[]) => mockCaptureException(...args),
+}));
+
 const mockWaitForPeerDbActivityDeletes = vi.fn().mockResolvedValue(undefined);
 const mockWaitForPeerDbActivityRestores = vi.fn().mockResolvedValue(undefined);
 const mockRunActivityReadModelBuild = vi.fn().mockResolvedValue(undefined);
 const mockInvalidateByPrefix = vi.fn().mockResolvedValue(undefined);
 const mockClose = vi.fn().mockResolvedValue(undefined);
 
+vi.mock("../logger.ts", () => ({
+  logger: { info: vi.fn(), warn: vi.fn() },
+}));
+
 vi.mock("../analytics/activity-read-model-build.ts", () => ({
   waitForPeerDbActivityDeletes: (...args: unknown[]) => mockWaitForPeerDbActivityDeletes(...args),
   waitForPeerDbActivityRestores: (...args: unknown[]) => mockWaitForPeerDbActivityRestores(...args),
@@ -27,13 +36,22 @@ vi.mock("../lib/cache.ts", () => ({
 
 import { processActivityDeleteAnalyticsJob } from "./process-activity-delete-analytics-job.ts";
 
-const job = {
-  data: {
-    type: "activity-delete-analytics-refresh" as const,
-    userId: "user-1",
-    activityIds: ["00000000-0000-0000-0000-000000000001"],
-  },
-};
+function createActivityAnalyticsJob(
+  type:
+    | "activity-delete-analytics-refresh"
+    | "activity-restore-analytics-refresh"
+    | "activity-recompute-analytics-refresh",
+  activityIds = ["00000000-0000-0000-0000-000000000001"],
+) {
+  return {
+    data: {
+      type,
+      userId: "user-1",
+      activityIds,
+    },
+    updateProgress: vi.fn().mockResolvedValue(undefined),
+  };
+}
 
 describe("processActivityDeleteAnalyticsJob", () => {
   beforeEach(() => {
@@ -45,9 +63,12 @@ describe("processActivityDeleteAnalyticsJob", () => {
     mockWaitForPeerDbActivityRestores.mockClear();
     mockWaitForPeerDbActivityRestores.mockResolvedValue(undefined);
     mockRunActivityReadModelBuild.mockResolvedValue(undefined);
+    mockCaptureException.mockClear();
   });
 
   it("waits for PeerDB, rebuilds activity read models, and invalidates user caches", async () => {
+    const job = createActivityAnalyticsJob("activity-delete-analytics-refresh");
+
     await processActivityDeleteAnalyticsJob(job);
 
     expect(mockWaitForPeerDbActivityDeletes).toHaveBeenCalledWith(expect.anything(), [
@@ -60,19 +81,18 @@ describe("processActivityDeleteAnalyticsJob", () => {
 
   it("closes the ClickHouse client even when refresh fails", async () => {
     mockRunActivityReadModelBuild.mockRejectedValueOnce(new Error("dbt failed"));
+    const job = createActivityAnalyticsJob("activity-delete-analytics-refresh");
 
     await expect(processActivityDeleteAnalyticsJob(job)).rejects.toThrow("dbt failed");
     expect(mockClose).toHaveBeenCalledOnce();
   });
 
   it("waits for restored activities before rebuilding read models", async () => {
-    await processActivityDeleteAnalyticsJob({
-      data: {
-        type: "activity-restore-analytics-refresh",
-        userId: "user-1",
-        activityIds: ["00000000-0000-0000-0000-000000000002"],
-      },
-    });
+    await processActivityDeleteAnalyticsJob(
+      createActivityAnalyticsJob("activity-restore-analytics-refresh", [
+        "00000000-0000-0000-0000-000000000002",
+      ]),
+    );
 
     expect(mockWaitForPeerDbActivityRestores).toHaveBeenCalledWith(expect.anything(), [
       "00000000-0000-0000-0000-000000000002",
@@ -84,18 +104,107 @@ describe("processActivityDeleteAnalyticsJob", () => {
   });
 
   it("rebuilds read models for recompute without waiting for PeerDB delete or restore state", async () => {
-    await processActivityDeleteAnalyticsJob({
-      data: {
-        type: "activity-recompute-analytics-refresh",
-        userId: "user-1",
-        activityIds: ["00000000-0000-0000-0000-000000000003"],
-      },
-    });
+    await processActivityDeleteAnalyticsJob(
+      createActivityAnalyticsJob("activity-recompute-analytics-refresh", [
+        "00000000-0000-0000-0000-000000000003",
+      ]),
+    );
 
     expect(mockWaitForPeerDbActivityDeletes).not.toHaveBeenCalled();
     expect(mockWaitForPeerDbActivityRestores).not.toHaveBeenCalled();
     expect(mockRunActivityReadModelBuild).toHaveBeenCalledOnce();
     expect(mockInvalidateByPrefix).toHaveBeenCalledWith("user-1:");
     expect(mockClose).toHaveBeenCalledOnce();
   });
+
+  it("reports delete analytics progress", async () => {
+    const job = createActivityAnalyticsJob("activity-delete-analytics-refresh");
+
+    await processActivityDeleteAnalyticsJob(job);
+
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 0,
+      message: "Starting activity analytics refresh...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 20,
+      message: "Waiting for activity deletes to reach analytics...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 60,
+      message: "Rebuilding activity analytics...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 90,
+      message: "Invalidating activity analytics cache...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 100,
+      message: "Activity analytics refresh complete.",
+    });
+  });
+
+  it("reports restore analytics progress", async () => {
+    const job = createActivityAnalyticsJob("activity-restore-analytics-refresh");
+
+    await processActivityDeleteAnalyticsJob(job);
+
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 0,
+      message: "Starting activity analytics refresh...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 20,
+      message: "Waiting for activity restores to reach analytics...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 60,
+      message: "Rebuilding activity analytics...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 90,
+      message: "Invalidating activity analytics cache...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 100,
+      message: "Activity analytics refresh complete.",
+    });
+  });
+
+  it("reports recompute analytics progress", async () => {
+    const job = createActivityAnalyticsJob("activity-recompute-analytics-refresh");
+
+    await processActivityDeleteAnalyticsJob(job);
+
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 0,
+      message: "Starting activity analytics refresh...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 60,
+      message: "Rebuilding activity analytics...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 90,
+      message: "Invalidating activity analytics cache...",
+    });
+    expect(job.updateProgress).toHaveBeenCalledWith({
+      percentage: 100,
+      message: "Activity analytics refresh complete.",
+    });
+  });
+
+  it("continues analytics refresh when progress updates fail", async () => {
+    const progressError = new Error("redis down");
+    const job = createActivityAnalyticsJob("activity-delete-analytics-refresh");
+    job.updateProgress = vi.fn().mockRejectedValue(progressError);
+
+    await processActivityDeleteAnalyticsJob(job);
+
+    expect(mockRunActivityReadModelBuild).toHaveBeenCalledOnce();
+    expect(mockInvalidateByPrefix).toHaveBeenCalledWith("user-1:");
+    expect(mockCaptureException).toHaveBeenCalledWith(progressError, {
+      tags: { activityAnalyticsStep: "updateProgress" },
+    });
+  });
 });

src/jobs/process-activity-delete-analytics-job.ts

@@ -1,3 +1,4 @@
+import * as Sentry from "@sentry/node";
 import {
   runActivityReadModelBuild,
   waitForPeerDbActivityDeletes,
@@ -10,36 +11,72 @@ import type { ActivityAnalyticsJobData } from "./queues.ts";
 
 export interface ActivityAnalyticsJob {
   data: ActivityAnalyticsJobData;
+  updateProgress: (data: { percentage: number; message: string }) => Promise<void>;
+}
+
+async function updateActivityAnalyticsProgress(
+  job: ActivityAnalyticsJob,
+  percentage: number,
+  message: string,
+): Promise<void> {
+  await job.updateProgress({ percentage, message }).catch((error: unknown) => {
+    logger.warn("Failed to update activity analytics progress: %s", error);
+    Sentry.captureException(error, { tags: { activityAnalyticsStep: "updateProgress" } });
+  });
+}
+
+async function rebuildActivityAnalytics(
+  job: ActivityAnalyticsJob,
+  userId: string,
+  logMessage: string,
+): Promise<void> {
+  await updateActivityAnalyticsProgress(job, 60, "Rebuilding activity analytics...");
+  await runActivityReadModelBuild();
+  await updateActivityAnalyticsProgress(job, 90, "Invalidating activity analytics cache...");
+  await queryCache.invalidateByPrefix(`${userId}:`);
+  logger.info(logMessage);
+  await updateActivityAnalyticsProgress(job, 100, "Activity analytics refresh complete.");
 }
 
 export async function processActivityDeleteAnalyticsJob(job: ActivityAnalyticsJob): Promise<void> {
   const { userId, activityIds } = job.data;
   const client = createClickHouseClientFromEnv();
 
   try {
+    await updateActivityAnalyticsProgress(job, 0, "Starting activity analytics refresh...");
     if (job.data.type === "activity-recompute-analytics-refresh") {
-      await runActivityReadModelBuild();
-      await queryCache.invalidateByPrefix(`${userId}:`);
-      logger.info(
+      await rebuildActivityAnalytics(
+        job,
+        userId,
         `[activity-recompute-analytics] Refreshed activity read models for ${activityIds.length} activities for user ${userId}`,
       );
       return;
     }
 
     if (job.data.type === "activity-restore-analytics-refresh") {
+      await updateActivityAnalyticsProgress(
+        job,
+        20,
+        "Waiting for activity restores to reach analytics...",
+      );
       await waitForPeerDbActivityRestores(client, activityIds);
-      await runActivityReadModelBuild();
-      await queryCache.invalidateByPrefix(`${userId}:`);
-      logger.info(
+      await rebuildActivityAnalytics(
+        job,
+        userId,
         `[activity-restore-analytics] Refreshed activity read models after restoring ${activityIds.length} activities for user ${userId}`,
       );
       return;
     }
 
+    await updateActivityAnalyticsProgress(
+      job,
+      20,
+      "Waiting for activity deletes to reach analytics...",
+    );
     await waitForPeerDbActivityDeletes(client, activityIds);
-    await runActivityReadModelBuild();
-    await queryCache.invalidateByPrefix(`${userId}:`);
-    logger.info(
+    await rebuildActivityAnalytics(
+      job,
+      userId,
       `[activity-delete-analytics] Refreshed activity read models after deleting ${activityIds.length} activities for user ${userId}`,
     );
   } finally {