eletrotupi / tcc/ api/src/lib/queue/processors/insights/index.tsd28d4c0c63d8f95f839b180d4753a0160e22f6bc
2.2 KB | 70 lines Raw
1
import { Job } from "bullmq";
2
import {
3
  InsightJobName,
4
  InsightJobData,
5
  InsightPeriodPayload,
6
} from "@app/lib/queue/types";
7
import { getQueue } from "@app/lib/queue/QueueRegistry";
8
import { prisma } from "@app/lib/prisma";
9
import { processMoodTrend } from "@app/lib/queue/processors/insights/moodTrend";
10
import { processEnergySleep } from "@app/lib/queue/processors/insights/energySleep";
11
import { processTriggerPattern } from "@app/lib/queue/processors/insights/triggerPattern";
12
import { processDailyEnergy } from "@app/lib/queue/processors/insights/dailyEnergy";
13
import { processDailySleep } from "@app/lib/queue/processors/insights/dailySleep";
14

15
export async function insightProcessor(
16
  job: Job<InsightJobData["data"]>,
17
): Promise<void> {
18
  console.log(`Insight job ${job.name} started`);
19

20
  switch (job.name as InsightJobName) {
21
    case InsightJobName.FanOut:
22
      return fanOut();
23

24
    case InsightJobName.MoodTrend:
25
      return processMoodTrend(job.data as InsightPeriodPayload);
26

27
    case InsightJobName.EnergySleep:
28
      return processEnergySleep(job.data as InsightPeriodPayload);
29

30
    case InsightJobName.TriggerPattern:
31
      return processTriggerPattern(job.data as InsightPeriodPayload);
32

33
    case InsightJobName.DailyEnergy:
34
      return processDailyEnergy(job.data as InsightPeriodPayload);
35

36
    case InsightJobName.DailySleep:
37
      return processDailySleep(job.data as InsightPeriodPayload);
38

39
    default:
40
      throw new Error(`Unknown insight job: ${job.name}`);
41
  }
42
}
43

44
async function fanOut(): Promise<void> {
45
  const users = await prisma.user.findMany({
46
    where: { active: true },
47
    select: { id: true },
48
  });
49

50
  const queue = getQueue("insights");
51
  const periodEnd = new Date();
52
  const periodStart = new Date();
53
  periodStart.setDate(periodEnd.getDate() - 7);
54

55
  const payload: Omit<InsightPeriodPayload, "userId"> = {
56
    periodStart: periodStart.toISOString(),
57
    periodEnd: periodEnd.toISOString(),
58
  };
59

60
  await queue.addBulk(
61
    users.flatMap((u) => [
62
      { name: InsightJobName.MoodTrend, data: { userId: u.id, ...payload } },
63
      { name: InsightJobName.EnergySleep, data: { userId: u.id, ...payload } },
64
      {
65
        name: InsightJobName.TriggerPattern,
66
        data: { userId: u.id, ...payload },
67
      },
68
    ]),
69
  );
70
}