Node.js AI Workflow with BullMQ: Reliable Tutorial

What You Will Build

This tutorial builds a small, production-minded work-intake service. Another system sends a work item to POST /webhooks/work-items. The API validates the payload, stores it in PostgreSQL, adds a BullMQ job, and returns 202 Accepted without waiting for an AI response.

A separate worker receives the job from Redis, loads the canonical record from PostgreSQL, asks OpenAI to classify the item, validates the returned JSON with Zod, and saves the outcome. The API exposes GET /work-items/:id for polling and GET /ready for dependency checks.

This separation is important. An LLM can assist with bounded interpretation such as classification and summarisation, but it should not become the system of record or the policy engine. PostgreSQL owns business state. Redis and BullMQ coordinate background execution. Application code enforces deterministic handling for security-sensitive categories.

The pattern is also relevant for GCC organisations that receive support, engineering, compliance, or operational requests across multiple systems. Before deploying, assess the data-residency, retention, Arabic-language evaluation, access-control, and regional hosting requirements that apply to your organisation.

Prerequisites

• Node.js 18 or later and npm.
• Docker Compose, or reachable PostgreSQL and Redis instances.
• An OpenAI API key and a model identifier available to your account.
• Basic TypeScript, SQL, HTTP, and environment-variable knowledge.

The verified workflow context supports the general architecture: AI orchestration systems use Redis-backed queues, background workers, APIs, task state, and external ticket providers. This tutorial deliberately keeps the stack self-managed and code-first. Teams that prefer managed TypeScript workflow infrastructure can evaluate that option separately, but the reliability boundaries described here still apply.

1. Create the Project

mkdir node-ai-workflow
cd node-ai-workflow
npm init -y
npm install bullmq dotenv express ioredis openai pg pino pino-http zod
npm install -D @types/express @types/node @types/pg tsx typescript
mkdir -p src db

Replace package.json with scripts for independent API and worker processes.

{
  "name": "node-ai-workflow",
  "private": true,
  "type": "module",
  "scripts": {
    "dev:api": "tsx watch src/api.ts",
    "dev:worker": "tsx watch src/worker.ts",
    "start:api": "tsx src/api.ts",
    "start:worker": "tsx src/worker.ts"
  }
}

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "outDir": "dist"
  },
  "include": ["src/**/*.ts"]
}

Create local PostgreSQL and Redis services. PostgreSQL is persistent workflow storage; Redis is the BullMQ processing dependency.

cat > docker-compose.yml <<'EOF'
services:
  postgres:
    image: postgres:alpine
    environment:
      POSTGRES_DB: ai_workflow
      POSTGRES_USER: workflow_user
      POSTGRES_PASSWORD: workflow_password
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
  redis:
    image: redis:alpine
    command: ["redis-server", "--appendonly", "yes"]
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
volumes:
  postgres_data:
  redis_data:
EOF

docker compose up -d

cat > .env <<'EOF'
PORT=3000
LOG_LEVEL=info
DATABASE_URL=postgresql://workflow_user:workflow_password@localhost:5432/ai_workflow
REDIS_URL=redis://localhost:6379
OPENAI_API_KEY=replace-with-your-key
OPENAI_MODEL=replace-with-a-model-available-to-your-account
WEBHOOK_SHARED_SECRET=local-development-secret-change-before-production
EOF

cat > .gitignore <<'EOF'
node_modules
dist
.env
*.log
EOF

2. Create the Database Schema

The unique idempotency_key is essential. Webhook senders can retry a delivery after a timeout or network failure. A unique database constraint turns duplicate delivery into a repeatable lookup instead of a second workflow run.

cat > db/001_create_work_items.sql <<'EOF'
CREATE EXTENSION IF NOT EXISTS pgcrypto;

CREATE TABLE work_items (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  idempotency_key TEXT NOT NULL UNIQUE,
  source TEXT NOT NULL,
  title TEXT NOT NULL,
  body TEXT NOT NULL,
  metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
  status TEXT NOT NULL DEFAULT 'queued'
    CHECK (status IN ('queued', 'processing', 'completed', 'failed')),
  attempt_count INTEGER NOT NULL DEFAULT 0,
  ai_result JSONB,
  failure_reason TEXT,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  completed_at TIMESTAMPTZ
);

CREATE INDEX work_items_status_created_at_idx
ON work_items (status, created_at DESC);
EOF

docker compose exec -T postgres psql -U workflow_user -d ai_workflow < db/001_create_work_items.sql

3. Add Shared Configuration and Schemas

cat > src/config.ts <<'EOF'
import "dotenv/config";
import { z } from "zod";const schema = z.object({
PORT: z.coerce.number().int().min(1).max(65535).default(3000),
LOG_LEVEL: z.enum(["trace", "debug", "info", "warn", "error", "fatal"]).default("info"),
DATABASE_URL: z.string().url(),
REDIS_URL: z.string().url(),
OPENAI_API_KEY: z.string().min(1),
OPENAI_MODEL: z.string().min(1),
WEBHOOK_SHARED_SECRET: z.string().min(16)
});const parsed = schema.safeParse(process.env);
if (!parsed.success) {
console.error(parsed.error.flatten().fieldErrors);
process.exit(1);
}
export const config = parsed.data;
EOFcat > src/db.ts <<'EOF'
import pg from "pg";
import { config } from "./config.js";
export const pool = new pg.Pool({ connectionString: config.DATABASE_URL, max: 10 });
EOFcat > src/redis.ts <<'EOF'
import IORedis from "ioredis";
import { config } from "./config.js";
export const redis = new IORedis(config.REDIS_URL, { maxRetriesPerRequest: null });
EOFcat > src/schemas.ts <<'EOF'
import { z } from "zod";export const inputSchema = z.object({
idempotencyKey: z.string().min(8).max(200),
source: z.string().min(2).max(100),
...
Continue Reading
Log in for free to read the rest of this article and access exclusive AI tools.
Log in / Register