Promises, setImmediate, and BullMQ

import { randomUUID } from 'node:crypto';
import { setImmediate as waitForImmediate } from 'node:timers/promises';
import {
  Queue,
  QueueEvents,
  Worker as BullWorker,
} from 'bullmq';

const sleep = (ms) =>
  new Promise((resolve) => setTimeout(resolve, ms));

function redisConnectionOptions(redisUrl) {
  const parsed = new URL(redisUrl);
  if (parsed.protocol !== 'redis:' && parsed.protocol !== 'rediss:') {
    throw new Error('REDIS_URL должен использовать redis:// или rediss://');
  }

  const database = Number.parseInt(parsed.pathname.slice(1) || '0', 10);
  return {
    host: parsed.hostname,
    port: Number.parseInt(parsed.port || '6379', 10),
    username: parsed.username || undefined,
    password: parsed.password || undefined,
    db: Number.isFinite(database) ? database : 0,
    tls: parsed.protocol === 'rediss:' ? {} : undefined,
    connectTimeout: 1200,
    maxRetriesPerRequest: null,
    enableOfflineQueue: false,
    retryStrategy: () => null,
  };
}

async function runBullMqRoundtrip(emit) {
  if (!process.env.REDIS_URL) {
    emit(
      'bullmq',
      'warning',
      'BullMQ-пример пропущен: задайте REDIS_URL или запустите проект через Docker Compose',
    );
    return;
  }

  const queueName = `node-loop-lab-${process.pid}-${randomUUID().slice(0, 8)}`;
  const connection = redisConnectionOptions(process.env.REDIS_URL);
  let queue;
  let worker;
  let queueEvents;
  const reportedConnectionErrors = new Set();
  const reportConnectionError = (source, error) => {
    const key = `${source}:${error.message}`;
    if (reportedConnectionErrors.has(key)) return;
    reportedConnectionErrors.add(key);
    emit(source, 'error', `${source} error: ${error.message}`);
  };

  try {
    queue = new Queue(queueName, {
      connection,
      defaultJobOptions: {
        attempts: 3,
        backoff: { type: 'exponential', delay: 100 },
        removeOnComplete: false,
        removeOnFail: false,
      },
    });
    queueEvents = new QueueEvents(queueName, { connection });
    worker = new BullWorker(
      queueName,
      async (job) => {
        emit(
          'bullmq-worker',
          'callback',
          `BullMQ Worker взял job ${job.id}: ${job.name}`,
        );
        await sleep(35);
        return {
          thumbnail: `${job.data.file}.webp`,
          processedBy: process.pid,
        };
      },
      { connection, concurrency: 1 },
    );

    queue.on('error', (error) => reportConnectionError('bullmq-queue', error));
    queueEvents.on('error', (error) =>
      reportConnectionError('bullmq-events', error),
    );
    worker.on('error', (error) =>
      reportConnectionError('bullmq-worker', error),
    );

    await Promise.all([
      queue.waitUntilReady(),
      queueEvents.waitUntilReady(),
      worker.waitUntilReady(),
    ]);

    emit(
      'bullmq-producer',
      'schedule',
      'Queue.add сохраняет job в Redis; HTTP-запрос не выполняет job сам',
    );
    const job = await queue.add('make-thumbnail', {
      file: 'avatar.png',
    });
    emit('redis', 'info', `Redis хранит job ${job.id} в состоянии waiting`);

    const result = await job.waitUntilFinished(queueEvents, 5000);
    emit(
      'bullmq-events',
      'result',
      `QueueEvents получил completed для job ${job.id}: ${result.thumbnail}`,
    );
  } catch (error) {
    emit(
      'bullmq',
      'warning',
      `BullMQ недоступен: ${error.message}. Promise-часть сценария уже выполнена.`,
    );
  } finally {
    await Promise.allSettled([worker?.close(), queueEvents?.close()]);
    if (queue) {
      try {
        await queue.obliterate({ force: true });
      } catch {
        // Redis мог стать недоступен во время очистки учебной очереди.
      }
      await queue.close().catch(() => {});
    }
  }
}

async function promisesImmediateBullMq(emit) {
  emit(
    'call-stack',
    'sync',
    'Создаём Promise: executor выполняется синхронно прямо сейчас',
  );

  const basePromise = new Promise((resolve) => {
    emit('promise-executor', 'sync', 'executor вызвал resolve(2)');
    resolve(2);
    emit(
      'promise-executor',
      'sync',
      'Код после resolve ещё выполняется, но повторно изменить outcome уже нельзя',
    );
  });

  const chainedPromise = basePromise
    .then((value) => {
      emit('microtasks', 'callback', `Первый then получил ${value} и вернул ${value * 3}`);
      return value * 3;
    })
    .then(async (value) => {
      await sleep(25);
      emit(
        'timers',
        'callback',
        `Второй then дождался Promise и получил ${value}`,
      );
      return value + 1;
    })
    .finally(() => {
      emit(
        'microtasks',
        'callback',
        'finally выполнился без подмены успешного результата',
      );
    });

  const recoveredPromise = Promise.reject(new Error('учебная ошибка'))
    .catch((error) => {
      emit('microtasks', 'warning', `catch обработал: ${error.message}`);
      return 'fallback';
    });

  const immediatePromise = new Promise((resolve) => {
    setImmediate(() => {
      emit(
        'check',
        'callback',
        'setImmediate callback выполняется в check-фазе',
      );
      resolve('immediate');
    });
  });

  emit(
    'call-stack',
    'schedule',
    'then/catch/setImmediate зарегистрированы; текущий стек освобождается',
  );
  const [chainResult, recovered] = await Promise.all([
    chainedPromise,
    recoveredPromise,
    immediatePromise,
  ]);
  emit(
    'result',
    'result',
    `Цепочка завершилась значением ${chainResult}; catch вернул ${recovered}`,
  );

  const aggregateTasks = [
    sleep(45).then(() => 'slow'),
    sleep(10).then(() => 'fast'),
    sleep(25).then(() => 'middle'),
  ];
  const allResult = await Promise.all(aggregateTasks);
  emit(
    'promise-all',
    'result',
    `Promise.all сохранил входной порядок: ${allResult.join(', ')}`,
  );

  const settled = await Promise.allSettled([
    Promise.resolve('ok'),
    Promise.reject(new Error('expected failure')),
  ]);
  emit(
    'promise-all-settled',
    'result',
    `Promise.allSettled вернул статусы: ${settled.map((item) => item.status).join(', ')}`,
  );

  const raceWinner = await Promise.race([
    sleep(30).then(() => 'timer 30ms'),
    waitForImmediate().then(() => 'setImmediate'),
  ]);
  emit('check', 'result', `Promise.race: первым завершился ${raceWinner}`);

  await waitForImmediate();
  emit(
    'check',
    'callback',
    'await setImmediate() из node:timers/promises продолжил функцию в check-фазе',
  );

  await runBullMqRoundtrip(emit);
}