schedule.ts · Source

schedule.tsapps/worker/src/processors/automations/schedule.tsraw ← automations6,642 B← Back to the app
1/**
2 * apps/worker — schedule triggers (TASK-0110 §4 + TASK-0113 schedule path).
3 *
4 * A BullMQ repeatable job per ACTIVE SCHEDULE automation, keyed
5 * `automations:run:{automationId}` (spec §4: the repeatable job id).
6 * `reconcileScheduleAutomations` is a set-difference over the queue's
7 * repeatables (idempotent — BullMQ upserts by name); it runs at boot and
8 * every 5 min (index.ts) so API-side activate/pause edits converge even if
9 * a reconcile was missed.
10 *
11 * When a repeatable fires, the worker handler calls `executeScheduleFire`:
12 * it creates a PENDING run with `triggerEvent = { type: "schedule",
13 * firedAt, cron, timeZone? }` (spec §4) and enqueues a `run-automation`
14 * job — the SAME pipeline as every other trigger.
15 */
16import type { PrismaClient } from "@prisma/client";
17import type { Redis } from "ioredis";
18import { Queue } from "bullmq";
19import { asJson } from "../../lib/prisma.js";
20import { getLogger } from "../../logger.js";
21import { QUEUES } from "../../queues.js";
22import { enqueueRunJob } from "./matcher.js";
23import { truncateJson } from "@factory/shared";
24import type { WorkerEnv } from "../../config.js";
25 
26const log = getLogger("automation-schedule");
27 
28/** Run `input` storage cap (spec: 64 KB). */
29const RUN_INPUT_MAX_BYTES = 64 * 1024;
30 
31/** The repeatable-job key for a schedule automation (spec §4). */
32export function scheduleJobKey(automationId: string): string {
33  return `automations:run:${automationId}`;
34}
35 
36/** True when a job name is one of ours (repeatable schedule fires). */
37export function isScheduleFireJobName(name: string): boolean {
38  return name.startsWith("automations:run:");
39}
40 
41export function cronFromConfig(config: unknown): string | null {
42  if (config && typeof config === "object") {
43    const c = config as Record<string, unknown>;
44    if (typeof c.cron === "string" && c.cron.trim()) return c.cron.trim();
45  }
46  return null;
47}
48 
49function timeZoneFromConfig(config: unknown): string | undefined {
50  if (config && typeof config === "object") {
51    const tz = (config as Record<string, unknown>).timeZone;
52    if (typeof tz === "string" && tz.trim()) return tz.trim();
53  }
54  return undefined;
55}
56 
57/**
58 * Register repeatable jobs for active SCHEDULE automations and remove stale
59 * ones (set difference by name — a shared literal name would make the
60 * removal loop delete every schedule on each reconcile). Idempotent.
61 */
62export async function reconcileScheduleAutomations(
63  connection: Redis,
64  prisma: PrismaClient
65): Promise<{ added: number; removed: number }> {
66  const q = new Queue(QUEUES.automations, { connection });
67  let added = 0;
68  let removed = 0;
69  try {
70    const automations = await prisma.automation.findMany({
71      where: { status: "ACTIVE" },
72      include: { triggers: { where: { type: "SCHEDULE" } } }
73    });
74    const wanted = new Map<string, { automationId: string; cron: string; tz?: string }>();
75    for (const a of automations) {
76      for (const t of a.triggers) {
77        const cron = cronFromConfig(t.config);
78        if (cron) {
79          const tz = timeZoneFromConfig(t.config);
80          wanted.set(scheduleJobKey(a.id), { automationId: a.id, cron, ...(tz ? { tz } : {}) });
81        }
82      }
83    }
84 
85    const existing = await q.getRepeatableJobs();
86    const existingNames = new Set(existing.map((r) => r.name));
87    for (const [name, spec] of wanted) {
88      if (!existingNames.has(name)) {
89        await q.add(
90          name,
91          { automationId: spec.automationId, event: "schedule", cron: spec.cron, ...(spec.tz ? { timeZone: spec.tz } : {}) },
92          {
93            repeat: { pattern: spec.cron, ...(spec.tz ? { tz: spec.tz } : {}) },
94            removeOnComplete: true,
95            removeOnFail: false
96          }
97        );
98        added++;
99        log.info({ automationId: spec.automationId, cron: spec.cron, tz: spec.tz ?? null }, "schedule registered");
100      }
101    }
102    for (const r of existing) {
103      if (!isScheduleFireJobName(r.name)) continue; // not ours
104      if (!wanted.has(r.name)) {
105        // Every one of our repeatables is added with a pattern (line above);
106        // BullMQ's removeRepeatable requires one — skip the unpatterned
107        // stragglers (another consumer's repeatable that happens to share
108        // the name prefix is not ours to remove).
109        if (!r.pattern) continue;
110        await q.removeRepeatable(r.name, { pattern: r.pattern });
111        removed++;
112        log.info({ name: r.name }, "schedule removed");
113      }
114    }
115    if (added || removed) log.info({ added, removed }, "schedule reconcile done");
116  } finally {
117    await q.close();
118  }
119  return { added, removed };
120}
121 
122/**
123 * A repeatable fired: create the PENDING run + enqueue the pipeline job.
124 * `data` is the repeatable's payload `{ automationId, event, cron, timeZone? }`.
125 * A deleted automation (row gone) is skipped — the next reconcile removes
126 * the repeatable.
127 */
128export async function executeScheduleFire(
129  connection: Redis,
130  prisma: PrismaClient,
131  env: WorkerEnv,
132  data: { automationId?: string; cron?: string; timeZone?: string }
133): Promise<{ runId: string } | null> {
134  void env;
135  if (!data.automationId) return null;
136  const automation = await prisma.automation.findUnique({
137    where: { id: data.automationId },
138    select: { id: true, workspaceId: true, status: true }
139  });
140  if (!automation || automation.status !== "ACTIVE") {
141    log.warn({ automationId: data.automationId }, "schedule fire for a missing/inactive automation (skipped)");
142    return null;
143  }
144 
145  const firedAt = new Date().toISOString();
146  const triggerEvent: Record<string, unknown> = {
147    type: "schedule",
148    firedAt,
149    cron: data.cron ?? null,
150    ...(data.timeZone ? { timeZone: data.timeZone } : {})
151  };
152  const { value: truncated, truncated: truncatedFlag } = truncateJson(triggerEvent, RUN_INPUT_MAX_BYTES);
153  const inputObj: Record<string, unknown> = {
154    triggerEvent: truncatedFlag ? { ...(truncated as Record<string, unknown>), truncated: true } : truncated,
155    conditions: null,
156    actionCount: 0
157  };
158 
159  const run = await prisma.automationRun.create({
160    data: {
161      workspaceId: automation.workspaceId,
162      automationId: automation.id,
163      status: "PENDING",
164      triggerEvent: asJson(triggerEvent),
165      input: asJson(inputObj),
166      retryCount: 0
167    }
168  });
169  await enqueueRunJob(connection, {
170    runId: run.id,
171    automationId: automation.id,
172    workspaceId: automation.workspaceId,
173    triggerEvent,
174    eventId: run.id
175  });
176  log.info({ runId: run.id, automationId: automation.id, cron: data.cron ?? null }, "schedule fire → run enqueued");
177  return { runId: run.id };
178}
179