feat(worker): BullMQ job queue for sequence steps and workflow delays

.gitignore

@@ -40,3 +40,6 @@ CLAUDE.md
 .claude/
 
 .vercel
+
+# node compile cache
+node-compile-cache

README.md

@@ -2,6 +2,17 @@
 
 Modern CRM with sequence automation, visual workflow builder, and a design system grounded in cognitive science research.
 
+## What this demonstrates
+
+The interesting part of a CRM is the automation engine, and getting it right means treating time and delivery as first-class problems. Relay runs sequences and workflows on a real BullMQ job queue backed by Redis, so a delay actually delays and an email actually fires when it should instead of blocking a request. The worker package leans on a few ideas worth calling out: at-least-once delivery with idempotent consumers (deterministic job IDs so a retried trigger can't double-schedule), delayed jobs for step timing, bounded retries with exponential backoff for transient failures like a flaky SMTP host, and backpressure through a fixed worker concurrency. Everything crossing a boundary is validated with Zod, and job routing goes through a small handler registry rather than a switch chain.
+
+## What's implemented
+
+- **packages/db** — Prisma schema for contacts, companies, deals, sequences, and workflows, with a seeded local database.
+- **packages/shared** — Zod schemas, workflow graph validation (cycle detection, topological sort, reachability), and sequence analytics.
+- **packages/web** — Next.js app: CRM views, sequence editor, and a React Flow workflow builder.
+- **packages/worker** — BullMQ queue on the declared Redis with two job types, `sequence-step` and `workflow-delay`, deterministic job IDs for idempotent scheduling, clamped delays, and a Zod-validated dispatcher.
+
 ## Tech Stack
 
 - **Framework:** Next.js 15+ (App Router, React 19)
@@ -20,6 +31,7 @@ relay/
 ├── packages/
 │   ├── web/        # Next.js frontend
 │   ├── shared/     # Zod schemas, types, constants
+│   ├── worker/     # BullMQ queue + background job processing
 │   └── db/         # Prisma schema, client, migrations
 ├── docker-compose.yml
 ├── turbo.json
@@ -69,6 +81,37 @@ pnpm test:integration
 pnpm test:e2e
 ```
 
+## Background Jobs
+
+The worker package schedules and runs sequence steps and workflow delays on Redis. Register a handler per job type, then start a worker.
+
+```ts
+import {
+  createConnection,
+  createQueue,
+  createWorker,
+  enqueueSequenceStep,
+  JobDispatcher,
+  JOB_NAMES,
+} from "@relay/worker";
+
+const connection = createConnection(); // reads REDIS_URL
+const queue = createQueue(connection);
+
+// Schedule the next step for an enrollment, 30 minutes out.
+await enqueueSequenceStep(queue, { enrollmentId, sequenceId, stepId }, 30 * 60 * 1000);
+
+const dispatcher = new JobDispatcher()
+  .on(JOB_NAMES.sequenceStep, async (payload) => {
+    /* send the email, advance the enrollment */
+  })
+  .on(JOB_NAMES.workflowDelay, async (payload) => {
+    /* resume the workflow execution from the delay node */
+  });
+
+createWorker(dispatcher, { connection });
+```
+
 ## Environment Variables
 
 Copy `.env.example` to `.env` and fill in the values. See the file for required variables.

docs/design-prompts/00-default.md

@@ -9,6 +9,7 @@ Paste the relevant section below into [claude.ai/design](https://claude.ai/desig
 > I'm working on `[PROJECT NAME]` — `[ONE-LINE DESCRIPTION OF WHAT IT IS, e.g. "an Electron desktop app for managing multi-project development workflows" or "a marketplace for vintage vinyl"]`. Stack: `[NEXT 15 / NUXT 4 / VITE + REACT 19 / SHOPIFY HYDROGEN / etc.]` with `[CSS MODULES / TAILWIND / CSS-IN-JS / VANILLA CSS]` for styling.
 >
 > The brand color palette uses these CSS variables (from `app/globals.css` or equivalent — replace with actual values):
+>
 > - `--color-accent: [HEX]` (primary brand)
 > - `--color-background: [HEX]` (page bg)
 > - `--color-foreground: [HEX]` (default text)
@@ -42,6 +43,7 @@ Paste the relevant section below into [claude.ai/design](https://claude.ai/desig
 ## Section 5 — Handoff prep (always include at end)
 
 > When the design is ready, package it as a Claude Code handoff bundle. I'll save the bundle to `design-bundles/<feature-slug>.json` in the project repo and pass it to Claude Code with: "implement design-bundles/<feature-slug>.json". Include in the bundle:
+>
 > - The component tree as nested elements with semantic HTML
 > - All CSS variables used (matching my project's existing names)
 > - Any new icons (preferring Lucide React icon names where they exist)

package.json

@@ -24,6 +24,7 @@
   },
   "devDependencies": {
     "prettier": "^3.4.2",
+    "prettier-plugin-tailwindcss": "^0.6.9",
     "turbo": "^2.3.3",
     "typescript": "^5.7.2"
   },

packages/worker/eslint.config.mjs

@@ -0,0 +1,36 @@
+import js from "@eslint/js";
+import tsPlugin from "@typescript-eslint/eslint-plugin";
+import tsParser from "@typescript-eslint/parser";
+
+export default [
+  js.configs.recommended,
+  {
+    files: ["src/**/*.ts"],
+    languageOptions: {
+      parser: tsParser,
+      parserOptions: {
+        ecmaVersion: 2022,
+        sourceType: "module",
+      },
+      globals: {
+        process: "readonly",
+        globalThis: "readonly",
+        console: "readonly",
+      },
+    },
+    plugins: {
+      "@typescript-eslint": tsPlugin,
+    },
+    rules: {
+      ...tsPlugin.configs.recommended.rules,
+      "@typescript-eslint/no-unused-vars": [
+        "error",
+        { argsIgnorePattern: "^_", varsIgnorePattern: "^_" },
+      ],
+      "no-unused-vars": "off",
+    },
+  },
+  {
+    ignores: ["dist/**", "node_modules/**"],
+  },
+];

packages/worker/package.json

@@ -0,0 +1,31 @@
+{
+  "name": "@relay/worker",
+  "version": "0.0.1",
+  "private": true,
+  "main": "./src/index.ts",
+  "types": "./src/index.ts",
+  "scripts": {
+    "build": "tsc",
+    "typecheck": "tsc --noEmit",
+    "lint": "eslint src/",
+    "lint:fix": "eslint src/ --fix",
+    "test": "vitest run",
+    "test:watch": "vitest",
+    "clean": "rm -rf dist"
+  },
+  "dependencies": {
+    "@relay/shared": "workspace:*",
+    "bullmq": "^5.34.10",
+    "ioredis": "^5.4.2",
+    "zod": "^3.24.1"
+  },
+  "devDependencies": {
+    "@eslint/js": "^9.39.2",
+    "@types/node": "^22.10.5",
+    "@typescript-eslint/eslint-plugin": "^8.56.0",
+    "@typescript-eslint/parser": "^8.56.0",
+    "eslint": "^9.18.0",
+    "typescript": "^5.7.2",
+    "vitest": "^2.1.8"
+  }
+}

packages/worker/src/dispatcher.test.ts

@@ -0,0 +1,60 @@
+import { describe, expect, it, vi } from "vitest";
+import { JobDispatcher, UnknownJobError } from "./dispatcher";
+import { JOB_NAMES } from "./payloads";
+
+const seqData = {
+  enrollmentId: "clh1a2b3c4d5e6f7g8h9i0j1",
+  sequenceId: "clh2a2b3c4d5e6f7g8h9i0j2",
+  stepId: "clh3a2b3c4d5e6f7g8h9i0j3",
+};
+
+describe("JobDispatcher", () => {
+  it("validates then routes to the registered handler", async () => {
+    const handler = vi.fn(async () => {});
+    const d = new JobDispatcher().on(JOB_NAMES.sequenceStep, handler);
+    await d.dispatch(JOB_NAMES.sequenceStep, seqData);
+    expect(handler).toHaveBeenCalledWith(seqData);
+  });
+
+  it("throws UnknownJobError for an unregistered known name", async () => {
+    const d = new JobDispatcher();
+    await expect(d.dispatch(JOB_NAMES.workflowDelay, {})).rejects.toBeInstanceOf(UnknownJobError);
+  });
+
+  it("throws UnknownJobError for a name it never knew", async () => {
+    const d = new JobDispatcher().on(JOB_NAMES.sequenceStep, async () => {});
+    await expect(d.dispatch("garbage", {})).rejects.toBeInstanceOf(UnknownJobError);
+  });
+
+  it("rejects an invalid payload without calling the handler", async () => {
+    const handler = vi.fn(async () => {});
+    const d = new JobDispatcher().on(JOB_NAMES.sequenceStep, handler);
+    await expect(d.dispatch(JOB_NAMES.sequenceStep, { enrollmentId: "x" })).rejects.toThrow();
+    expect(handler).not.toHaveBeenCalled();
+  });
+
+  it("propagates a handler failure so the queue can retry", async () => {
+    const d = new JobDispatcher().on(JOB_NAMES.sequenceStep, async () => {
+      throw new Error("smtp down");
+    });
+    await expect(d.dispatch(JOB_NAMES.sequenceStep, seqData)).rejects.toThrow("smtp down");
+  });
+
+  it("reports which names it can handle", () => {
+    const d = new JobDispatcher().on(JOB_NAMES.workflowDelay, async () => {});
+    expect(d.handles(JOB_NAMES.workflowDelay)).toBe(true);
+    expect(d.handles(JOB_NAMES.sequenceStep)).toBe(false);
+    expect(d.handles("nope")).toBe(false);
+  });
+
+  it("keeps the last handler registered for a name", async () => {
+    const first = vi.fn(async () => {});
+    const second = vi.fn(async () => {});
+    const d = new JobDispatcher()
+      .on(JOB_NAMES.sequenceStep, first)
+      .on(JOB_NAMES.sequenceStep, second);
+    await d.dispatch(JOB_NAMES.sequenceStep, seqData);
+    expect(first).not.toHaveBeenCalled();
+    expect(second).toHaveBeenCalledOnce();
+  });
+});

packages/worker/src/dispatcher.ts

@@ -0,0 +1,35 @@
+import { JOB_PAYLOAD_SCHEMAS, isJobName, type JobName, type JobPayloads } from "./payloads";
+
+export class UnknownJobError extends Error {
+  constructor(public readonly jobName: string) {
+    super(`No handler registered for job "${jobName}"`);
+    this.name = "UnknownJobError";
+  }
+}
+
+export type JobHandler<N extends JobName> = (payload: JobPayloads[N]) => Promise<void>;
+
+type AnyHandler = (payload: unknown) => Promise<void>;
+
+export class JobDispatcher {
+  private readonly handlers = new Map<JobName, AnyHandler>();
+
+  on<N extends JobName>(name: N, handler: JobHandler<N>): this {
+    this.handlers.set(name, handler as AnyHandler);
+    return this;
+  }
+
+  handles(name: string): name is JobName {
+    return isJobName(name) && this.handlers.has(name);
+  }
+
+  // Throws on unknown name or invalid payload so BullMQ marks the job failed and
+  // retries it under the queue's backoff policy rather than silently dropping it.
+  async dispatch(name: string, data: unknown): Promise<void> {
+    if (!isJobName(name)) throw new UnknownJobError(name);
+    const handler = this.handlers.get(name);
+    if (!handler) throw new UnknownJobError(name);
+    const payload = JOB_PAYLOAD_SCHEMAS[name].parse(data);
+    await handler(payload);
+  }
+}

packages/worker/src/index.ts

@@ -0,0 +1,23 @@
+import { Worker, type Job } from "bullmq";
+import type { Redis } from "ioredis";
+import { JobDispatcher } from "./dispatcher";
+import { QUEUE_NAME } from "./queue";
+
+export * from "./payloads";
+export * from "./queue";
+export * from "./dispatcher";
+
+export interface WorkerOptions {
+  connection: Redis;
+  concurrency?: number;
+}
+
+export function createWorker(dispatcher: JobDispatcher, opts: WorkerOptions): Worker {
+  return new Worker(
+    QUEUE_NAME,
+    async (job: Job) => {
+      await dispatcher.dispatch(job.name, job.data);
+    },
+    { connection: opts.connection, concurrency: opts.concurrency ?? 5 },
+  );
+}

packages/worker/src/payloads.test.ts

@@ -0,0 +1,64 @@
+import { describe, expect, it } from "vitest";
+import {
+  isJobName,
+  JOB_NAMES,
+  JOB_PAYLOAD_SCHEMAS,
+  sequenceStepPayloadSchema,
+  workflowDelayPayloadSchema,
+} from "./payloads";
+
+const CUID_A = "clh1a2b3c4d5e6f7g8h9i0j1";
+const CUID_B = "clh2a2b3c4d5e6f7g8h9i0j2";
+const CUID_C = "clh3a2b3c4d5e6f7g8h9i0j3";
+
+describe("payload schemas", () => {
+  it("accepts a well-formed sequence-step payload", () => {
+    const parsed = sequenceStepPayloadSchema.parse({
+      enrollmentId: CUID_A,
+      sequenceId: CUID_B,
+      stepId: CUID_C,
+    });
+    expect(parsed.stepId).toBe(CUID_C);
+  });
+
+  it("rejects non-cuid ids", () => {
+    expect(() =>
+      sequenceStepPayloadSchema.parse({ enrollmentId: "1", sequenceId: CUID_B, stepId: CUID_C }),
+    ).toThrow();
+  });
+
+  it("strips unknown keys on workflow-delay", () => {
+    const parsed = workflowDelayPayloadSchema.parse({
+      executionId: CUID_A,
+      workflowId: CUID_B,
+      nodeId: CUID_C,
+      injected: "nope",
+    });
+    expect(parsed).not.toHaveProperty("injected");
+  });
+
+  it("rejects a payload missing a required field", () => {
+    expect(() =>
+      workflowDelayPayloadSchema.parse({ executionId: CUID_A, workflowId: CUID_B }),
+    ).toThrow();
+  });
+});
+
+describe("isJobName", () => {
+  it("recognizes the two known job names", () => {
+    expect(isJobName(JOB_NAMES.sequenceStep)).toBe(true);
+    expect(isJobName(JOB_NAMES.workflowDelay)).toBe(true);
+  });
+
+  it("rejects anything else", () => {
+    expect(isJobName("send-slack")).toBe(false);
+    expect(isJobName("")).toBe(false);
+  });
+});
+
+describe("JOB_PAYLOAD_SCHEMAS", () => {
+  it("maps each name to its schema", () => {
+    expect(JOB_PAYLOAD_SCHEMAS[JOB_NAMES.sequenceStep]).toBe(sequenceStepPayloadSchema);
+    expect(JOB_PAYLOAD_SCHEMAS[JOB_NAMES.workflowDelay]).toBe(workflowDelayPayloadSchema);
+  });
+});