feat: create process-week queue for improved week processing (#225)

Author: Alex Stelea

apps/admin/src/app/seasons/[seasonId]/weeks/[weekId]/page.tsx

@@ -15,8 +15,7 @@ type WeekPageProps = {
 
 const WeekPage: FC<WeekPageProps> = ({ params: paramsPromise }) => {
   const params = use(paramsPromise);
-  const recalculatePoints =
-    api.season.addCalculateSeasonPointsJob.useMutation();
+  const addProcessWeekJob = api.season.addProcessWeekJob.useMutation();
   const updatePointsPool = api.week.updatePointsPool.useMutation();
   const updateMultiplier = api.week.updateActivityWeekMultiplier.useMutation();
   const {
@@ -51,9 +50,8 @@ const WeekPage: FC<WeekPageProps> = ({ params: paramsPromise }) => {
   const week = weekData;
 
   const handleProcessWeek = async () => {
-    await recalculatePoints.mutateAsync({
+    await addProcessWeekJob.mutateAsync({
       weekId: params.weekId,
-      force: week?.processed,
     });
 
     toast.info('Processing week job started', {

apps/workers/src/queues/process-week/queue.ts

@@ -0,0 +1,13 @@
+import { redisClient } from '../../redis';
+import { createQueue } from '../createQueue';
+import { QueueName } from '../types';
+import { processWeekWorker } from './worker';
+
+export const processWeekQueue = createQueue({
+  name: QueueName.processWeek,
+  redisClient,
+  worker: processWeekWorker,
+  onError: async (_, error) => {
+    console.error(error);
+  },
+});

apps/workers/src/queues/process-week/schemas.ts

@@ -0,0 +1,7 @@
+import { z } from 'zod';
+
+export const ProcessWeekJobSchema = z.object({
+  weekId: z.string(),
+});
+
+export type ProcessWeekJob = z.infer<typeof ProcessWeekJobSchema>;

apps/workers/src/queues/process-week/worker.ts

@@ -0,0 +1,61 @@
+import { dependencyLayer } from 'api/incentives';
+import type { Job } from 'bullmq';
+import { FlowProducer } from 'bullmq';
+import { Exit } from 'effect';
+import { handleExitError } from '../../helpers/handleExitError';
+import { redisClient } from '../../redis';
+import { QueueName } from '../types';
+import type { ProcessWeekJob } from './schemas';
+
+const flowProducer = new FlowProducer({ connection: redisClient });
+
+export const processWeekWorker = async (job: Job<ProcessWeekJob>) => {
+  const weekId = job.data.weekId;
+
+  const seasonResult = await dependencyLayer.getSeasonByWeekId(weekId);
+
+  if (Exit.isFailure(seasonResult)) {
+    return handleExitError(seasonResult);
+  }
+
+  const seasonId = seasonResult.value.id;
+
+  job.log(`starting scheduled calculations for weekId: ${weekId}`);
+
+  await flowProducer.add({
+    queueName: QueueName.populateLeaderboardCache,
+    name: 'process-week',
+    data: { weekId },
+    opts: {
+      attempts: 3,
+      backoff: {
+        type: 'exponential',
+        delay: 5000,
+      },
+    },
+    children: [
+      {
+        queueName: QueueName.calculateSeasonPoints,
+        name: 'process-week',
+        data: { weekId, seasonId, markAsProcessed: true },
+        opts: { failParentOnFailure: true },
+        children: [
+          {
+            queueName: QueueName.seasonPointsMultiplier,
+            name: 'process-week',
+            data: { weekId },
+            opts: { failParentOnFailure: true },
+            children: [
+              {
+                queueName: QueueName.calculateActivityPoints,
+                name: 'process-week',
+                data: { weekId },
+                opts: { failParentOnFailure: true },
+              },
+            ],
+          },
+        ],
+      },
+    ],
+  });
+};

apps/workers/src/queues/scheduled-calculations/worker.ts

@@ -4,7 +4,7 @@ import { FlowProducer } from 'bullmq';
 import { Exit } from 'effect';
 import { handleExitError } from '../../helpers/handleExitError';
 import { redisClient } from '../../redis';
-import { calculateSeasonPointsQueue } from '../calculate-season-points/queue';
+import { processWeekQueue } from '../process-week/queue';
 import { QueueName } from '../types';
 import type { ScheduledCalculationsJob } from './schemas';
 
@@ -41,9 +41,8 @@ export const scheduledCalculationsWorker = async (
   if (unprocessedWeeks.length > 0) {
     for (const week of unprocessedWeeks) {
       job.log(`adding end-of-week-calculation job for weekId: ${week.weekId}`);
-      await calculateSeasonPointsQueue.queue.add('end-of-week-calculation', {
+      await processWeekQueue.queue.add('process-week', {
         weekId: week.weekId,
-        markAsProcessed: true,
       });
     }
   }

apps/workers/src/queues/types.ts

@@ -8,6 +8,7 @@ export const QueueName = {
   snapshotDateRange: 'snapshotDateRange',
   scheduledCalculations: 'scheduledCalculations',
   populateLeaderboardCache: 'populateLeaderboardCache',
+  processWeek: 'processWeek',
 } as const;
 
 export type QueueName = (typeof QueueName)[keyof typeof QueueName];

apps/workers/src/server/index.ts

@@ -19,6 +19,8 @@ import { eventQueue } from '../queues/event/queue';
 import { eventQueueJobSchema } from '../queues/event/schemas';
 import { populateLeaderboardCacheQueue } from '../queues/populate-leaderboard-cache/queue';
 import { populateLeaderboardCacheSchema } from '../queues/populate-leaderboard-cache/schemas';
+import { processWeekQueue } from '../queues/process-week/queue';
+import { ProcessWeekJobSchema } from '../queues/process-week/schemas';
 import { scheduledCalculationsQueue } from '../queues/scheduled-calculations/queue';
 import { scheduledSnapshotQueue } from '../queues/scheduled-snapshot/queue';
 import { snapshotQueue } from '../queues/snapshot/queue';
@@ -46,6 +48,8 @@ metricsApp.get('/metrics', async (c) => {
     await scheduledCalculationsQueue.queue.exportPrometheusMetrics();
   const populateLeaderboardCacheQueueMetrics =
     await populateLeaderboardCacheQueue.queue.exportPrometheusMetrics();
+  const processWeekQueueMetrics =
+    await processWeekQueue.queue.exportPrometheusMetrics();
   return c.text(
     [
       snapshotQueueMetrics,
@@ -55,6 +59,7 @@ metricsApp.get('/metrics', async (c) => {
       calculateActivityPointsQueueMetrics,
       scheduledCalculationsQueueMetrics,
       populateLeaderboardCacheQueueMetrics,
+      processWeekQueueMetrics,
     ].join('\n'),
   );
 });
@@ -116,6 +121,16 @@ app.post('/queues/calculate-season-points/add', async (c) => {
   return c.text('ok');
 });
 
+app.post('/queues/process-week/add', async (c) => {
+  const input = await c.req.json();
+  const parsedInput = ProcessWeekJobSchema.safeParse(input);
+  if (!parsedInput.success) {
+    return c.json({ error: parsedInput.error.message }, 400);
+  }
+  await processWeekQueue.queue.add('process-week', parsedInput.data);
+  return c.text('ok');
+});
+
 app.post('/queues/calculate-season-points-multiplier/add', async (c) => {
   const input = await c.req.json();
   const parsedInput = seasonPointsMultiplierJobSchema.safeParse(input);
@@ -169,6 +184,7 @@ createBullBoard({
     new BullMQAdapter(seasonPointsMultiplierQueue.queue),
     new BullMQAdapter(scheduledCalculationsQueue.queue),
     new BullMQAdapter(populateLeaderboardCacheQueue.queue),
+    new BullMQAdapter(processWeekQueue.queue),
   ],
   serverAdapter,
 });

packages/api/src/incentives/season/seasonRouter.ts

@@ -145,23 +145,19 @@ export const adminSeasonRouter = createTRPCRouter({
       });
     }),
 
-  addCalculateSeasonPointsJob: publicProcedure
+  addProcessWeekJob: publicProcedure
     .input(
       z.object({
         weekId: z.string(),
-        force: z.boolean().optional(),
       }),
     )
     .mutation(async ({ input }) => {
       const response = await fetch(
-        `${process.env.WORKERS_API_BASE_URL}/queues/scheduled-calculations/add`,
+        `${process.env.WORKERS_API_BASE_URL}/queues/process-week/add`,
         {
           method: 'POST',
           body: JSON.stringify({
             weekId: input.weekId,
-            force: input.force,
-            markAsProcessed: true,
-            includeSPCalculations: true,
           }),
         },
       );