Background Processing

Worker Threads

Offload CPU-intensive operations from the main event loop using Node.js worker threads. Perfect for JWT operations, password hashing, encryption, and heavy computations.

Overview

Good Use Cases

JWT token signing/verification
Password hashing (bcrypt, argon2)
Data encryption/decryption
Large data transformations
Image/video processing
Data compression/decompression

Not Recommended

Simple I/O operations (use async/await)
Database queries (already non-blocking)
Network requests (already non-blocking)
Small computations (overhead > benefit)

Getting Started

Basic Usagetypescript

1import { createApp, getWorkerManager } from '@morojs/moro';
2
3const app = await createApp();
4const workers = getWorkerManager();
5
6// Execute a task on worker thread
7app.post('/api/hash', async (req, res) => {
8  const result = await workers.executeTask({
9    id: `hash-${Date.now()}`,
10    type: 'crypto:hash',
11    data: {
12      input: req.body.password,
13      algorithm: 'sha256'
14    }
15  });
16
17  return { hash: result };
18});
19
20app.listen(3000);

Configurationtypescript

1import { createApp, getWorkerManager, WorkerManager } from '@morojs/moro';
2
3// Configure the pool at app startup — initialized eagerly and
4// shut down automatically on app.close()
5const app = await createApp({
6  workers: {
7    count: 4, // Number of worker threads (default: CPU cores - 1)
8    maxQueueSize: 1000 // Maximum queued tasks (default: 1000)
9  }
10});
11
12// Or configure the shared singleton directly (created on first call)
13const workers = getWorkerManager({
14  workerCount: 4,
15  maxQueueSize: 1000
16});
17
18// Or manage your own instance
19const dedicated = new WorkerManager({
20  workerCount: 2,
21  maxQueueSize: 500
22});

Built-in Tasks

JWT Operationstypescript

1import { workerTasks } from '@morojs/moro';
2
3// Verify JWT token (returns the decoded payload, throws if invalid)
4app.post('/api/auth/verify', async (req, res) => {
5  try {
6    const payload = await workerTasks.verifyJWT(
7      req.body.token,
8      process.env.JWT_SECRET
9    );
10    return { user: payload };
11  } catch {
12    return res.status(401).json({ error: 'Invalid token' });
13  }
14});
15
16// Sign JWT token
17app.post('/api/auth/login', async (req, res) => {
18  const user = await authenticateUser(req.body);
19
20  const token = await workerTasks.signJWT(
21    { userId: user.id, role: user.role },
22    process.env.JWT_SECRET,
23    { expiresIn: '7d' }
24  );
25
26  return { token };
27});

Hashingtypescript

1// Hash data on a worker thread
2app.post('/api/register', async (req, res) => {
3  const hash = await workerTasks.hash(
4    req.body.password,
5    'sha256'
6  );
7
8  const user = await createUser({
9    email: req.body.email,
10    password: hash
11  });
12
13  return user;
14});

Data Compressiontypescript

1// Compress large payloads off the main thread
2// Formats: 'gzip' (default), 'deflate', 'brotli'
3app.get('/api/export', async (req, res) => {
4  const data = await getLargeDataset();
5
6  const compressed = await workerTasks.compress(
7    JSON.stringify(data),
8    { format: 'gzip' }
9  );
10
11  res.setHeader('Content-Encoding', 'gzip');
12  res.setHeader('Content-Type', 'application/json');
13  return compressed;
14});
15
16// Decompress incoming data
17app.post('/api/import', async (req, res) => {
18  const decompressed = await workerTasks.decompress(
19    req.body.data,
20    { format: 'gzip' }
21  );
22
23  const data = JSON.parse(decompressed.toString('utf8'));
24  await importData(data);
25
26  return { success: true, records: data.length };
27});

Heavy Computation & JSON Transformstypescript

1// Offload heavy computation
2app.post('/api/analyze', async (req, res) => {
3  const result = await workerTasks.heavyComputation({
4    operation: 'analysis',
5    dataset: req.body.dataset
6  });
7
8  return { result };
9});
10
11// Transform large JSON payloads off the main thread
12app.post('/api/import', async (req, res) => {
13  const transformed = await workerTasks.transformJSON(
14    req.body.records,
15    data => data.filter(row => row.active)
16  );
17
18  await importData(transformed);
19  return { success: true, records: transformed.length };
20});

Task Priority & Timeout

Task Prioritytypescript

1import { getWorkerManager } from '@morojs/moro';
2
3const workers = getWorkerManager();
4
5// High priority task (executes first)
6await workers.executeTask({
7  id: 'critical-task',
8  type: 'crypto:hash',
9  data: { input: 'critical-data' },
10  priority: 'high' // 'high' | 'normal' | 'low'
11});
12
13// Normal priority (default)
14await workers.executeTask({
15  id: 'normal-task',
16  type: 'crypto:hash',
17  data: { input: 'normal-data' },
18  priority: 'normal'
19});
20
21// Low priority (executes last)
22await workers.executeTask({
23  id: 'background-task',
24  type: 'crypto:hash',
25  data: { input: 'background-data' },
26  priority: 'low'
27});

Task Timeouttypescript

1await workers.executeTask({
2  id: 'long-task',
3  type: 'computation:heavy',
4  data: { operation: 'complex-calculation' },
5  timeout: 30000 // 30 seconds
6});

Best Practices

1. Use Workers for CPU-Intensive Tasks Only

typescript

1// ✅ Good: CPU-intensive
2await workerTasks.hash(password, 'sha256');
3await workerTasks.compress(largePayload, { format: 'brotli' });
4await workerTasks.transformJSON(largeDataset, transformer);
5await workerTasks.heavyComputation(complexCalc);
6
7// ❌ Bad: I/O operations (already non-blocking)
8await db.users.find(); // Use normal async/await
9await fetch('https://api.example.com'); // Already non-blocking

2. Monitor Worker Health

typescript

1// Check worker stats periodically
2// getStats() returns { workerCount, activeTasks, queuedTasks, isShuttingDown }
3setInterval(() => {
4  const stats = workers.getStats();
5
6  if (stats.queuedTasks > 500) {
7    console.warn('Worker queue is getting large:', stats);
8  }
9
10  if (stats.activeTasks === stats.workerCount) {
11    console.log('All workers are busy');
12  }
13}, 10000);
14
15// Expose metrics endpoint
16app.get('/metrics/workers', (req, res) => {
17  return workers.getStats();
18});

3. Graceful Shutdown

typescript

1import { getWorkerManager } from '@morojs/moro';
2
3const workers = getWorkerManager();
4
5process.on('SIGTERM', async () => {
6  console.log('Shutting down workers...');
7  await workers.shutdown();
8  process.exit(0);
9});

Next Steps

Authentication

Use worker threads for JWT operations

Performance

Optimize your application performance

API Reference

Complete API documentation