NestJS Setting up a node:worker_thread

etl.worker.entry.ts

import { parentPort, workerData } from 'node:worker_threads';
import { NestFactory } from '@nestjs/core';
import { EtlWorkerModule } from './etl.worker.module';
import { EtlWorkerService } from './etl.worker.service';

(async () => {
  if (!parentPort) throw new Error('Must be run as a worker thread');

  const app = await NestFactory.createApplicationContext(EtlWorkerModule, {
    logger: ['error', 'warn'],
  });

  try {
    const svc = app.get(EtlWorkerService);
    const result = await svc.process(workerData as { rows: number[] });
    parentPort.postMessage({ ok: true, result });
  } catch (err) {
    parentPort.postMessage({ ok: false, error: (err as Error).message });
  } finally {
    await app.close();
  }
})();

etl.worker.module.ts

import { Module } from '@nestjs/common';
import { EtlWorkerService } from './etl.worker.service';

@Module({
  providers: [EtlWorkerService],
  exports: [EtlWorkerService],
})
export class EtlWorkerModule {}

etl.worker.service.ts

import { Injectable, Logger } from '@nestjs/common';

@Injectable()
export class EtlWorkerService {
  private readonly logger = new Logger(EtlWorkerService.name);

  async process(payload: { rows: number[] }) {
    this.logger.log(`Processing ${payload.rows.length} rows`);
    // CPU-ish work goes here
    const data = payload.rows.map((n) => n * 2);
    return { count: data.length, data };
  }
}

worker-manager.service.ts

import { Injectable, Logger } from '@nestjs/common';
import { Worker } from 'node:worker_threads';
import { join } from 'node:path';

@Injectable()
export class WorkerManagerService {
  private readonly logger = new Logger(WorkerManagerService.name);

  runEtl(rows: number[]): Promise<{ count: number; data: number[] }> {
    // points to the compiled JS in dist/
    const workerPath = join(__dirname, 'workers', 'etl.worker.entry.js');

    return new Promise((resolve, reject) => {
      const worker = new Worker(workerPath, { workerData: { rows } });

      worker.once('message', (msg: any) => {
        if (msg?.ok) resolve(msg.result);
        else reject(new Error(msg?.error ?? 'Worker failed'));
      });

      worker.once('error', (err) => {
        this.logger.error(err);
        reject(err);
      });

      worker.once('exit', (code) => {
        if (code !== 0) this.logger.warn(`Worker exited with code ${code}`);
      });
    });
  }
}