Фоновые задачи и очереди: что выносить из запроса

npm init -y && npm pkg set type=module
npm install bullmq ioredis express
npm install -D typescript @types/express @types/node
docker run -d --rm --name kb-redis-jobs -p 6379:6379 redis:7-alpine

{
  "compilerOptions": {
    "target": "es2024",
    "module": "nodenext",
    "strict": true,
    "noEmit": true,
    "allowImportingTsExtensions": true,
    "erasableSyntaxOnly": true,
    "verbatimModuleSyntax": true,
    "skipLibCheck": true
  },
  "include": ["src"]
}

import { Redis } from 'ioredis';

const url = process.env.REDIS_URL ?? 'redis://localhost:6379';

// The producer runs inside an HTTP handler: if Redis is down,
// fail fast instead of holding the client's request.
export const producerConnection = () =>
  new Redis(url, { maxRetriesPerRequest: 1, enableOfflineQueue: false });

// The worker must ride out disconnects: BullMQ requires null here.
export const workerConnection = () => new Redis(url, { maxRetriesPerRequest: null });

import { Queue } from 'bullmq';
import type { Redis } from 'ioredis';

export type LeadNotification = { leadId: string; name: string; phone: string };

export const NOTIFICATIONS = 'notifications';

export function createNotificationsQueue(connection: Redis) {
  return new Queue<LeadNotification>(NOTIFICATIONS, {
    connection,
    defaultJobOptions: {
      attempts: 4, // first try + 3 retries
      backoff: { type: 'exponential', delay: 1000 }, // waits 1 s, 2 s, 4 s
      removeOnComplete: { count: 1000 }, // keep the last 1000 successful jobs
      removeOnFail: { age: 7 * 24 * 3600 }, // keep failed jobs for a week
    },
  });
}

import { randomUUID } from 'node:crypto';
import express from 'express';
import { createNotificationsQueue } from './queue.ts';
import { producerConnection } from './redis.ts';

type Lead = { id: string; name: string; phone: string; createdAt: string };

const leads = new Map<string, Lead>(); // a database table in a real project
const queue = createNotificationsQueue(producerConnection());
queue.on('error', (err) => console.error('[queue]', err.message || err.name));

// While Redis is down, queue.add waits for a reconnect: don't let the request hang.
async function withTimeout<T>(promise: Promise<T>, ms: number): Promise<T> {
  let timer: NodeJS.Timeout | undefined;
  const timeout = new Promise<never>((_, reject) => {
    timer = setTimeout(() => reject(new Error(`timeout after ${ms} ms`)), ms);
  });
  promise.catch(() => {}); // a late rejection must not crash the process
  try {
    return await Promise.race([promise, timeout]);
  } finally {
    clearTimeout(timer);
  }
}

const app = express();
app.use(express.json());

app.post('/api/leads', async (req, res) => {
  const { name, phone } = req.body ?? {};
  if (typeof name !== 'string' || typeof phone !== 'string') {
    res.status(400).json({ error: 'name and phone are required' });
    return;
  }

  // The client generates the key once per form submission:
  // a repeated request creates neither a second lead nor a second job.
  const id = req.get('Idempotency-Key') ?? randomUUID();
  if (leads.has(id)) {
    res.status(200).json({ id, duplicate: true });
    return;
  }
  leads.set(id, { id, name, phone, createdAt: new Date().toISOString() });

  try {
    // jobId dedupes at the queue level: a job with an existing id is ignored
    const job = queue.add('lead-created', { leadId: id, name, phone }, { jobId: `lead-${id}` });
    await withTimeout(job, 2000);
  } catch (err) {
    // The lead is saved but the notification is not queued: make it visible in logs.
    console.error('[api] enqueue failed', id, (err as Error).message);
  }

  res.status(201).json({ id }); // the response does not wait for the notification
});

const port = Number(process.env.PORT ?? 3000);
app.listen(port, () => console.log(`[api] http://localhost:${port}`));

$ curl -s -X POST localhost:3000/api/leads … -w '  HTTP %{http_code}, %{time_total}s\n'
{"id":"ab6a5c91-26cb-4c2a-aac4-f5fe5842624c"}  HTTP 201, 2.019723s

# api
[api] enqueue failed ab6a5c91-26cb-4c2a-aac4-f5fe5842624c timeout after 2000 ms

import { UnrecoverableError, Worker, type Job } from 'bullmq';
import { Redis } from 'ioredis';
import { NOTIFICATIONS, type LeadNotification } from './queue.ts';
import { workerConnection } from './redis.ts';

const NOTIFY_URL = process.env.NOTIFY_URL ?? 'http://localhost:4001/notify';
const store = new Redis(process.env.REDIS_URL ?? 'redis://localhost:6379'); // "already sent" marks
const log = (msg: string) => console.log(new Date().toISOString().slice(11, 19), msg);

async function notifyManager(job: Job<LeadNotification>): Promise<string> {
  const doneKey = `lead-notified:${job.data.leadId}`;
  // A retry after a successful send (e.g. the worker died before completing the job)
  if (await store.exists(doneKey)) return 'already sent';

  const res = await fetch(NOTIFY_URL, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify({ text: `New lead: ${job.data.name}, ${job.data.phone}` }),
    signal: AbortSignal.timeout(5000), // a hung request is an error, not an endless wait
  });

  // A 4xx (except 429) won't be fixed by retrying: fail now, skip the remaining attempts
  if (res.status >= 400 && res.status < 500 && res.status !== 429) {
    throw new UnrecoverableError(`notify rejected: HTTP ${res.status}`);
  }
  if (!res.ok) throw new Error(`notify unavailable: HTTP ${res.status}`);

  await store.set(doneKey, '1', 'EX', 7 * 24 * 3600);
  return 'sent';
}

const worker = new Worker<LeadNotification, string>(NOTIFICATIONS, notifyManager, {
  connection: workerConnection(),
  concurrency: 5,
});

worker.on('completed', (job, result) => {
  log(`${job.id} ${result} (attempt ${job.attemptsMade})`);
});

worker.on('failed', async (job, err) => {
  if (!job) return;
  if ((await job.getState()) === 'failed') {
    // Out of attempts or unrecoverable: a dead job that needs a human
    log(`DEAD ${job.id} after ${job.attemptsMade} attempt(s): ${err.message}`);
  } else {
    log(`retry ${job.id} attempt ${job.attemptsMade} failed: ${err.message}`);
  }
});

worker.on('error', (err) => console.error('[worker]', err.message));

async function shutdown() {
  log('shutting down');
  await worker.close(); // finish active jobs, take no new ones
  await store.quit();
  process.exit(0);
}
process.on('SIGINT', shutdown);
process.on('SIGTERM', shutdown);

log('waiting for jobs');

// A local stand-in for the notification service (Telegram, email, CRM).
import { createServer } from 'node:http';

let failFirst = Number(process.env.FAIL_FIRST ?? 2); // how many first requests fail

createServer(async (req, res) => {
  let body = '';
  for await (const chunk of req) body += chunk;
  if (failFirst > 0) {
    failFirst--;
    console.log('[notify] 503 Service Unavailable');
    res.writeHead(503).end();
    return;
  }
  const { text } = JSON.parse(body);
  if (!/\d{10,}/.test(text)) {
    console.log('[notify] 400 Bad Request');
    res.writeHead(400).end();
    return;
  }
  console.log(`[notify] 200 delivered: ${text}`);
  res.writeHead(200).end();
}).listen(4001, () => console.log('[notify] http://localhost:4001'));

FAIL_FIRST=2 node src/fake-notify.ts   # terminal 1: the "notification service"
node src/worker.ts                      # terminal 2: the worker
node src/api.ts                         # terminal 3: the API on :3000

$ curl -s -X POST localhost:3000/api/leads -H 'content-type: application/json' \
    -H 'Idempotency-Key: 7f3c2a10' --data-binary @lead.json -w '  HTTP %{http_code}, %{time_total}s\n'
{"id":"7f3c2a10"}  HTTP 201, 0.015741s
$ curl …the same command again…
{"id":"7f3c2a10","duplicate":true}  HTTP 200, 0.004159s

# worker
23:24:53 waiting for jobs
23:24:55 retry lead-7f3c2a10 attempt 1 failed: notify unavailable: HTTP 503
23:24:56 retry lead-7f3c2a10 attempt 2 failed: notify unavailable: HTTP 503
23:24:58 lead-7f3c2a10 sent (attempt 3)

# fake-notify
[notify] 503 Service Unavailable
[notify] 503 Service Unavailable
[notify] 200 delivered: New lead: Aigerim, +77011234567

$ curl -s -X POST localhost:3000/api/leads -H 'content-type: application/json' \
    -H 'Idempotency-Key: 9b1e44d0' -d '{"name":"Test","phone":"none"}'
{"id":"9b1e44d0"}

# worker
23:25:03 DEAD lead-9b1e44d0 after 1 attempt(s): notify rejected: HTTP 400

$ node check/replay.ts   # the same leadId under a new jobId

# worker
23:25:28 lead-7f3c2a10-replay already sent (attempt 1)

# fake-notify: no new requests

# fake-notify started with FAIL_FIRST=100: the service is "down"
23:25:09 retry lead-c5d2e9a1 attempt 1 failed: notify unavailable: HTTP 503
23:25:10 retry lead-c5d2e9a1 attempt 2 failed: notify unavailable: HTTP 503
23:25:12 retry lead-c5d2e9a1 attempt 3 failed: notify unavailable: HTTP 503
23:25:16 DEAD lead-c5d2e9a1 after 4 attempt(s): notify unavailable: HTTP 503

$ node src/dead-jobs.ts
lead-c5d2e9a1  attempts=4  reason="notify unavailable: HTTP 503"
lead-9b1e44d0  attempts=1  reason="notify rejected: HTTP 400"
failed: 2

import { createNotificationsQueue } from './queue.ts';
import { producerConnection } from './redis.ts';

const connection = producerConnection();
const queue = createNotificationsQueue(connection);

const failed = await queue.getFailed();
for (const job of failed) {
  console.log(`${job.id}  attempts=${job.attemptsMade}  reason="${job.failedReason}"`);
  if (process.argv.includes('--retry')) await job.retry(); // back to the queue after a fix
}
console.log(`failed: ${failed.length}`);

await queue.close();
await connection.quit(); // we created this connection, so we close it

# the service is back (FAIL_FIRST=0)
$ node src/dead-jobs.ts --retry
lead-c5d2e9a1  attempts=4  reason="notify unavailable: HTTP 503"
lead-9b1e44d0  attempts=1  reason="notify rejected: HTTP 400"
failed: 2

# worker
23:25:21 lead-c5d2e9a1 sent (attempt 5)
23:25:21 DEAD lead-9b1e44d0 after 2 attempt(s): notify rejected: HTTP 400

$ docker stop <container with node src/worker.ts>
23:25:56 waiting for jobs
23:25:57 shutting down
container exit code: 0

import { Queue } from 'bullmq';
import { producerConnection } from './redis.ts';

const connection = producerConnection();
const reports = new Queue('reports', { connection });

// Repeatable job: a lead summary every day at 09:00 Almaty time.
// upsert: running this on every deploy updates the schedule instead of duplicating it.
await reports.upsertJobScheduler(
  'daily-leads-report',
  { pattern: '0 0 9 * * *', tz: 'Asia/Almaty' },
  { name: 'daily-leads-report', data: {} },
);

// Delayed job: remind the manager if nobody picks up the lead within 30 minutes.
await reports.add('lead-reminder', { leadId: '7f3c2a10' }, { delay: 30 * 60 * 1000 });

for (const s of await reports.getJobSchedulers()) {
  console.log(s.key, s.pattern, s.tz, 'next:', new Date(s.next ?? 0).toISOString());
}
console.log(await reports.getJobCounts('delayed'));

await reports.close();
await connection.quit();

$ node src/schedule.ts
daily-leads-report 0 0 9 * * * Asia/Almaty next: 2026-10-12T04:00:00.000Z
{ delayed: 2 }