Node.js

Node.js Architecture:

┌───────────────────────────────────────────────────┐
│                   JavaScript                       │
├───────────────────────────────────────────────────┤
│                      Node.js                       │
├─────────────┬───────────────────────┬─────────────┤
│   Node API  │        V8 Engine      │    libuv    │
└─────────────┴───────────────────────┴─────────────┘
        

Event Loop Phases in Detail:

/**
 * Node.js Event Loop Phases:
 * 1. timers: executes setTimeout() and setInterval() callbacks
 * 2. pending callbacks: executes I/O callbacks deferred to the next loop iteration
 * 3. idle, prepare: used internally by Node.js
 * 4. poll: retrieves new I/O events; executes I/O related callbacks
 * 5. check: executes setImmediate() callbacks
 * 6. close callbacks: executes close event callbacks like socket.on('close', ...)
 */

// This demonstrates the event loop phases
console.log('1: Program start');

setTimeout(() => console.log('2: Timer phase'), 0);

setImmediate(() => console.log('3: Check phase'));

process.nextTick(() => console.log('4: Next tick (runs before phases start)'));

Promise.resolve().then(() => console.log('5: Promise (microtask queue)'));

// Simulating an I/O operation
fs.readFile(__filename, () => {
  console.log('6: I/O callback (poll phase)');
  
  setTimeout(() => console.log('7: Nested timer'), 0);
  setImmediate(() => console.log('8: Nested immediate (prioritized after I/O)'));
  process.nextTick(() => console.log('9: Nested next tick'));
});

console.log('10: Program end');

// Output order demonstrates event loop phases and priorities
        

Example: Module caching behavior

// counter.js
let count = 0;
module.exports = {
  increment: function() {
    return ++count;
  },
  getCount: function() {
    return count;
  }
};

// app.js
const counter1 = require('./counter');
const counter2 = require('./counter');

console.log(counter1.increment()); // 1
console.log(counter2.increment()); // 2 (not 1, because the module is cached)
console.log(counter1 === counter2); // true
        

Example: Using ES Modules in Node.js

// math.mjs or package.json with "type": "module"
export function add(a, b) {
  return a + b;
}

export function subtract(a, b) {
  return a - b;
}

// main.mjs
import { add, subtract } from './math.mjs';
console.log(add(5, 3)); // 8
        

Version Specifiers in package.json:

"dependencies": {
  "express": "4.17.1",       // Exact version
  "lodash": "^4.17.21",      // Compatible with 4.17.21 up to < 5.0.0
  "moment": "~2.29.1",       // Compatible with 2.29.1 up to < 2.30.0
  "webpack": ">=5.0.0",      // Version 5.0.0 or higher
  "react": "16.x",           // Any 16.x.x version
  "typescript": "*"          // Any version
}
        

npm Publishing Workflow:

# Login to npm
npm login

# Bump version (updates package.json)
npm version patch|minor|major

# Publish to registry
npm publish
        

Node.js I/O Architecture:

┌─────────────────────────────┐
│       Node.js Process       │
│                             │
│  ┌─────────┐   ┌─────────┐  │
│  │   JS    │   │  Event  │  │
│  │  Code   │═══▶  Loop   │  │
│  └─────────┘   └────┬────┘  │
│                     │       │
│  ┌─────────┐   ┌────▼────┐  │
│  │  Sync   │   │  libuv  │  │
│  │   I/O   │◄──┤ Thread  │  │
│  │ Binding │   │  Pool   │  │
│  └─────────┘   └─────────┘  │
└─────────────────────────────┘
        

Thread Pool Configuration Impact:

// The default thread pool size is 4
// You can increase it for better I/O parallelism
process.env.UV_THREADPOOL_SIZE = 8;

// Now Node.js can handle 8 concurrent file operations
// without degrading performance

// Measuring the impact
const fs = require('fs');
const files = Array(16).fill('large_file.txt');

console.time('parallel-io');
let completed = 0;

files.forEach((file, index) => {
  fs.readFile(file, (err, data) => {
    completed++;
    console.log(`Completed ${completed} of ${files.length}`);
    
    if (completed === files.length) {
      console.timeEnd('parallel-io');
    }
  });
});
        

Comprehensive HTTP Server Implementation:

const http = require('http');
const url = require('url');

// Server creation with detailed request handler
const server = http.createServer((req, res) => {
  // Parse the request URL
  const parsedUrl = url.parse(req.url, true);
  const path = parsedUrl.pathname;
  const trimmedPath = path.replace(/^\/+|\/+$/g, '');
  
  // Get the request method, headers, and query string parameters
  const method = req.method.toLowerCase();
  const headers = req.headers;
  const queryStringObject = parsedUrl.query;
  
  // Collect request body data if present
  let buffer = [];
  req.on('data', (chunk) => {
    buffer.push(chunk);
  });
  
  // Process the complete request once all data is received
  req.on('end', () => {
    buffer = Buffer.concat(buffer).toString();
    
    // Prepare response object
    const responseData = {
      trimmedPath,
      method,
      headers,
      queryStringObject,
      payload: buffer ? JSON.parse(buffer) : {}
    };
    
    // Log request information
    console.log(`Request received: ${method.toUpperCase()} ${trimmedPath}`);
    
    // Set response headers
    res.setHeader('Content-Type', 'application/json');
    
    // Send response
    res.writeHead(200);
    res.end(JSON.stringify(responseData));
  });
});

// Configure server with error handling and IPv6 dual-stack support
const PORT = process.env.PORT || 3000;
server.listen(PORT, () => {
  console.log(`Server running on port ${PORT}`);
}).on('error', (err) => {
  console.error(`Server error: ${err.message}`);
});
        

Efficient Body Parsing with Streams:

const http = require('http');

// Handle potentially large payloads efficiently using streams
const server = http.createServer((req, res) => {
  // Stream validation setup
  const contentLength = parseInt(req.headers['content-length'] || '0');
  
  if (contentLength > 10_000_000) { // 10MB limit
    res.writeHead(413, {'Content-Type': 'text/plain'});
    res.end('Payload too large');
    req.destroy(); // Terminate the connection
    return;
  }
  
  // Error handling for the request stream
  req.on('error', (err) => {
    console.error('Request stream error:', err);
    res.statusCode = 400;
    res.end('Bad Request');
  });

  // Using stream processing for data collection
  if (req.method === 'POST' || req.method === 'PUT') {
    const chunks = [];
    
    req.on('data', (chunk) => {
      chunks.push(chunk);
    });
    
    req.on('end', () => {
      try {
        // Process the complete payload
        const rawBody = Buffer.concat(chunks);
        let body;
        
        const contentType = req.headers['content-type'] || '';
        
        if (contentType.includes('application/json')) {
          body = JSON.parse(rawBody.toString());
        } else if (contentType.includes('application/x-www-form-urlencoded')) {
          body = new URLSearchParams(rawBody.toString());
        } else {
          body = rawBody; // Raw buffer for binary data
        }
        
        // Continue with request processing
        processRequest(req, res, body);
      } catch (error) {
        console.error('Error processing request body:', error);
        res.statusCode = 400;
        res.end('Invalid request payload');
      }
    });
  } else {
    // Handle non-body requests (GET, DELETE, etc.)
    processRequest(req, res);
  }
});

function processRequest(req, res, body) {
  // Application logic here...
}

Optimized HTTP Response Management:

const http = require('http');
const fs = require('fs');
const path = require('path');
const zlib = require('zlib');

const server = http.createServer((req, res) => {
  // Handle compression based on Accept-Encoding
  const acceptEncoding = req.headers['accept-encoding'] || '';
  
  // Response helpers
  function sendJSON(data, statusCode = 200) {
    // Optimizes buffering with cork/uncork
    res.cork();
    res.setHeader('Content-Type', 'application/json');
    res.statusCode = statusCode;
    
    // Prepare JSON response
    const jsonStr = JSON.stringify(data);
    
    // Apply compression if supported
    if (acceptEncoding.includes('br')) {
      res.setHeader('Content-Encoding', 'br');
      const compressed = zlib.brotliCompressSync(jsonStr);
      res.setHeader('Content-Length', compressed.length);
      res.end(compressed);
    } else if (acceptEncoding.includes('gzip')) {
      res.setHeader('Content-Encoding', 'gzip');
      const compressed = zlib.gzipSync(jsonStr);
      res.setHeader('Content-Length', compressed.length);
      res.end(compressed);
    } else {
      res.setHeader('Content-Length', Buffer.byteLength(jsonStr));
      res.end(jsonStr);
    }
    res.uncork();
  }
  
  function sendFile(filePath, contentType) {
    const fullPath = path.join(__dirname, filePath);
    
    // File access error handling
    fs.access(fullPath, fs.constants.R_OK, (err) => {
      if (err) {
        res.statusCode = 404;
        res.end('File not found');
        return;
      }
      
      // Stream the file with proper headers
      res.setHeader('Content-Type', contentType);
      
      // Add caching headers for static assets
      res.setHeader('Cache-Control', 'max-age=86400'); // 1 day
      
      // Streaming with compression for text-based files
      if (contentType.includes('text/') || 
          contentType.includes('application/javascript') ||
          contentType.includes('application/json') || 
          contentType.includes('xml')) {
          
        const fileStream = fs.createReadStream(fullPath);
        
        if (acceptEncoding.includes('gzip')) {
          res.setHeader('Content-Encoding', 'gzip');
          fileStream.pipe(zlib.createGzip()).pipe(res);
        } else {
          fileStream.pipe(res);
        }
      } else {
        // Stream binary files directly
        fs.createReadStream(fullPath).pipe(res);
      }
    });
  }
  
  // Route handling logic with the helpers
  if (req.url === '/api/data' && req.method === 'GET') {
    sendJSON({ message: 'Success', data: [1, 2, 3] });
  } else if (req.url === '/styles.css') {
    sendFile('public/styles.css', 'text/css');
  } else {
    // Handle other routes...
  }
});

HTTP/2 Server Example:

const http2 = require('http2');
const fs = require('fs');

const server = http2.createSecureServer({
  key: fs.readFileSync('key.pem'),
  cert: fs.readFileSync('cert.pem')
});

server.on('stream', (stream, headers) => {
  // HTTP/2 uses streams instead of req/res
  const path = headers[':path'];
  
  if (path === '/') {
    stream.respond({
      'content-type': 'text/html',
      ':status': 200
    });
    stream.end('<h1>HTTP/2 Server</h1>');
  } else if (path === '/resource') {
    // Server push example
    stream.pushStream({ ':path': '/style.css' }, (err, pushStream) => {
      if (err) throw err;
      pushStream.respond({ ':status': 200, 'content-type': 'text/css' });
      pushStream.end('body { color: red; }');
    });
    
    stream.respond({ ':status': 200 });
    stream.end('Resource with pushed CSS');
  }
});

server.listen(443);

Try-Catch Block Anatomy:

try {
  // Code that might throw an error
  const result = riskyOperation();
  processResult(result);
} catch (error) {
  // Error handling logic
  if (error instanceof TypeError) {
    // Handle type errors specially
  } else if (error instanceof RangeError) {
    // Handle range errors
  } else {
    // Generic error handling
  }
} finally {
  // Optional block that always executes
  // Used for cleanup operations
  releaseResources();
}
        

Error-First Callback Implementation:

// Consuming an error-first callback API
fs.readFile("/path/to/file", (err, data) => {
  if (err) {
    // Early return pattern for error handling
    return handleError(err);
  }
  
  // Success path
  processData(data);
});

// Implementing a function that accepts an error-first callback
function readConfig(filename, callback) {
  fs.readFile(filename, (err, data) => {
    if (err) {
      // Propagate the error to the caller
      return callback(err);
    }
    
    try {
      // Note: Synchronous errors inside callbacks should be caught
      // and passed to the callback
      const config = JSON.parse(data);
      callback(null, config);
    } catch (parseError) {
      callback(parseError);
    }
  });
}
        

Promise Implementation (Simplified):

class MyPromise {
  constructor(executor) {
    this.state = "pending";
    this.value = undefined;
    this.reason = undefined;
    this.onFulfilledCallbacks = [];
    this.onRejectedCallbacks = [];

    try {
      executor(
        // resolve function
        (value) => {
          if (this.state === "pending") {
            this.state = "fulfilled";
            this.value = value;
            this.onFulfilledCallbacks.forEach(cb => cb(this.value));
          }
        }, 
        // reject function
        (reason) => {
          if (this.state === "pending") {
            this.state = "rejected";
            this.reason = reason;
            this.onRejectedCallbacks.forEach(cb => cb(this.reason));
          }
        }
      );
    } catch (error) {
      if (this.state === "pending") {
        this.state = "rejected";
        this.reason = error;
        this.onRejectedCallbacks.forEach(cb => cb(this.reason));
      }
    }
  }

  then(onFulfilled, onRejected) {
    // Implementation of .then() with proper promise chaining...
  }

  catch(onRejected) {
    return this.then(null, onRejected);
  }
}
        

Event Loop and Promises:

console.log("Start");

setTimeout(() => {
  console.log("Timeout callback");
}, 0);

Promise.resolve().then(() => {
  console.log("Promise callback");
});

console.log("End");

// Output:
// Start
// End
// Promise callback
// Timeout callback
        

Promise Composition:

// Promise.all - waits for all promises to resolve or any to reject
Promise.all([fetchUser(1), fetchUser(2), fetchUser(3)])
  .then(users => { /* all users available */ })
  .catch(error => { /* any error from any promise */ });

// Promise.race - resolves/rejects as soon as any promise resolves/rejects
Promise.race([
  fetch("/resource"),
  new Promise((_, reject) => setTimeout(() => reject(new Error("Timeout")), 5000))
])
  .then(response => { /* handle response */ })
  .catch(error => { /* handle error or timeout */ });

// Promise.allSettled - waits for all promises to settle (fulfill or reject)
Promise.allSettled([fetchUser(1), fetchUser(2), fetchUser(3)])
  .then(results => {
    // results is an array of objects with status and value/reason
    results.forEach(result => {
      if (result.status === "fulfilled") {
        console.log("Success:", result.value);
      } else {
        console.log("Error:", result.reason);
      }
    });
  });

// Promise.any - resolves when any promise resolves, rejects only if all reject
Promise.any([fetchData(1), fetchData(2), fetchData(3)])
  .then(firstSuccess => { /* use first successful result */ })
  .catch(aggregateError => { /* all promises failed */ });
        

Conceptual Implementation of Async/Await:

// This is a simplified conceptual model of how async/await works internally
function asyncFunction(generatorFunction) {
  return function(...args) {
    const generator = generatorFunction(...args);
    
    return new Promise((resolve, reject) => {
      function step(method, arg) {
        try {
          const result = generator[method](arg);
          const { value, done } = result;
          
          if (done) {
            resolve(value);
          } else {
            Promise.resolve(value)
              .then(val => step("next", val))
              .catch(err => step("throw", err));
          }
        } catch (error) {
          reject(error);
        }
      }
      
      step("next", undefined);
    });
  };
}

// The async function:
// async function foo() {
//   const result = await somePromise;
//   return result + 1;
// }

// Would be transformed to something like:
const foo = asyncFunction(function* () {
  const result = yield somePromise;
  return result + 1;
});
        

Sequential vs Concurrent Execution:

// Sequential execution - slower when operations are independent
async function sequential() {
  console.time("sequential");
  
  const result1 = await operation1(); // Wait for this to finish
  const result2 = await operation2(); // Then start this
  const result3 = await operation3(); // Then start this
  
  console.timeEnd("sequential");
  return [result1, result2, result3];
}

// Concurrent execution - faster for independent operations
async function concurrent() {
  console.time("concurrent");
  
  // Start all operations immediately
  const promise1 = operation1();
  const promise2 = operation2();
  const promise3 = operation3();
  
  // Then wait for all to complete
  const result1 = await promise1;
  const result2 = await promise2;
  const result3 = await promise3;
  
  console.timeEnd("concurrent");
  return [result1, result2, result3];
}

// Even more concise with Promise.all
async function concurrentWithPromiseAll() {
  console.time("promise.all");
  
  const results = await Promise.all([
    operation1(),
    operation2(),
    operation3()
  ]);
  
  console.timeEnd("promise.all");
  return results;
}
        

Error Handling with Async/Await:

// Pattern 1: Using try/catch with specific error types
async function errorHandlingWithTypes() {
  try {
    const data = await fetchData();
    return processData(data);
  } catch (error) {
    if (error instanceof NetworkError) {
      // Handle network errors
      await reconnect();
      return errorHandlingWithTypes(); // Retry
    } else if (error instanceof ValidationError) {
      // Handle validation errors
      return { error: "Invalid data format", details: error.details };
    } else {
      // Log unexpected errors
      console.error("Unexpected error:", error);
      throw error; // Re-throw for upstream handling
    }
  }
}

// Pattern 2: Higher-order function for retry logic
const withRetry = (fn, maxRetries = 3, delay = 1000) => async (...args) => {
  let lastError;
  
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await fn(...args);
    } catch (error) {
      console.warn(`Attempt ${attempt + 1} failed:`, error);
      lastError = error;
      
      if (attempt < maxRetries - 1) {
        await new Promise(resolve => setTimeout(resolve, delay * (attempt + 1)));
      }
    }
  }
  
  throw new Error(`Failed after ${maxRetries} attempts. Last error: ${lastError}`);
};

// Usage
const reliableFetch = withRetry(fetchData);
const data = await reliableFetch(url);

// Pattern 3: Error boundary pattern
async function errorBoundary(asyncFn) {
  try {
    return { data: await asyncFn(), error: null };
  } catch (error) {
    return { data: null, error };
  }
}

// Usage
const { data, error } = await errorBoundary(() => fetchUserData(userId));
if (error) {
  // Handle error case
} else {
  // Use data
}
        

Advanced Stream Implementation Example:

const { Transform } = require('stream');
const fs = require('fs');
const zlib = require('zlib');

// Create a custom Transform stream for data processing
class CustomTransformer extends Transform {
  constructor(options = {}) {
    super(options);
    this.totalProcessed = 0;
  }

  _transform(chunk, encoding, callback) {
    // Process the data chunk (convert to uppercase in this example)
    const transformedChunk = chunk.toString().toUpperCase();
    this.totalProcessed += chunk.length;
    
    // Push the transformed data to the output buffer
    this.push(transformedChunk);
    
    // Signal that the transformation is complete
    callback();
  }

  _flush(callback) {
    // Add metadata at the end of the stream
    this.push(`\nProcessed ${this.totalProcessed} bytes total`);
    callback();
  }
}

// Create a streaming pipeline with backpressure handling
fs.createReadStream('input.txt')
  .pipe(new CustomTransformer())
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('output.txt.gz'))
  .on('finish', () => console.log('Pipeline processing complete'))
  .on('error', (err) => console.error('Pipeline error', err));
        

1. Readable Streams

Readable streams implement the stream.Readable interface and operate in one of two modes:

• Flowing mode: Data is pushed from the source as soon as it arrives
• Paused mode: Data must be explicitly requested using the read() method

Core implementation requirements include:

• Implementing the _read(size) method that pushes data to the internal buffer
• Managing the highWaterMark to control buffering behavior
• Proper state management between flowing/paused modes and error states

const { Readable } = require('stream');

class TimeStream extends Readable {
  constructor(options = {}) {
    // Merge options with defaults
    super({ objectMode: true, ...options });
    this.startTime = Date.now();
    this.maxReadings = options.maxReadings || 10;
    this.count = 0;
  }

  _read() {
    if (this.count >= this.maxReadings) {
      this.push(null); // Signal end of stream
      return;
    }

    // Simulate async data production with throttling
    setTimeout(() => {
      try {
        const reading = {
          timestamp: Date.now(),
          elapsed: Date.now() - this.startTime,
          readingNumber: ++this.count
        };
        
        // Push the reading into the buffer
        this.push(reading);
      } catch (err) {
        this.emit('error', err);
      }
    }, 100);
  }
}

// Usage
const timeData = new TimeStream({ maxReadings: 5 });
timeData.on('data', data => console.log(data));
timeData.on('end', () => console.log('Stream complete'));
        

2. Writable Streams

Writable streams implement the stream.Writable interface and provide a destination for data.

Core implementation considerations:

• Implementing the _write(chunk, encoding, callback) method that handles data consumption
• Optional implementation of _writev(chunks, callback) for optimized batch writing
• Buffer management with highWaterMark to handle backpressure
• State tracking for pending writes, corking, and drain events

const { Writable } = require('stream');
const fs = require('fs');

class DatabaseWriteStream extends Writable {
  constructor(options = {}) {
    super({ objectMode: true, ...options });
    this.db = options.db || null;
    this.batchSize = options.batchSize || 100;
    this.buffer = [];
    this.totalWritten = 0;
    
    // Create a log file for failed writes
    this.errorLog = fs.createWriteStream('db-write-errors.log', { flags: 'a' });
  }

  _write(chunk, encoding, callback) {
    if (!this.db) {
      process.nextTick(() => callback(new Error('Database not connected')));
      return;
    }

    // Add to buffer
    this.buffer.push(chunk);
    
    // Flush if we've reached batch size
    if (this.buffer.length >= this.batchSize) {
      this._flushBuffer(callback);
    } else {
      // Continue immediately
      callback();
    }
  }
  
  _final(callback) {
    // Flush any remaining items in buffer
    if (this.buffer.length > 0) {
      this._flushBuffer(callback);
    } else {
      callback();
    }
  }
  
  _flushBuffer(callback) {
    const batchToWrite = [...this.buffer];
    this.buffer = [];
    
    // Mock DB write operation
    this.db.batchWrite(batchToWrite, (err, result) => {
      if (err) {
        // Log errors but don't fail the stream - retry logic could be implemented here
        this.errorLog.write(JSON.stringify({ 
          time: new Date(), 
          error: err.message,
          failedBatchSize: batchToWrite.length
        }) + '\n');
      } else {
        this.totalWritten += result.inserted;
      }
      callback();
    });
  }
}
        

3. Duplex Streams

Duplex streams implement both Readable and Writable interfaces, providing bidirectional data flow.

Implementation requirements:

• Implementing both _read(size) and _write(chunk, encoding, callback) methods
• Maintaining separate internal buffer states for reading and writing
• Properly handling events for both interfaces (drain, data, end, finish)

const { Duplex } = require('stream');

class ProtocolBridge extends Duplex {
  constructor(options = {}) {
    super(options);
    this.sourceProtocol = options.sourceProtocol;
    this.targetProtocol = options.targetProtocol;
    this.conversionState = { 
      pendingRequests: new Map(),
      maxPending: options.maxPending || 100
    };
  }

  _read(size) {
    // Pull response data from target protocol
    this.targetProtocol.getResponses(size, (err, responses) => {
      if (err) {
        this.emit('error', err);
        return;
      }
      
      // Process each response and push to readable side
      for (const response of responses) {
        // Match with pending request from mapping table
        const originalRequest = this.conversionState.pendingRequests.get(response.id);
        if (originalRequest) {
          // Convert response format back to source protocol format
          const convertedResponse = this._convertResponseFormat(response, originalRequest);
          this.push(convertedResponse);
          
          // Remove from pending tracking
          this.conversionState.pendingRequests.delete(response.id);
        }
      }
      
      // If no responses and read buffer getting low, push some empty padding
      if (responses.length === 0 && this.readableLength < size/2) {
        this.push(Buffer.alloc(0)); // Empty buffer, keeps stream active
      }
    });
  }

  _write(chunk, encoding, callback) {
    // Convert source protocol format to target protocol format
    try {
      const request = JSON.parse(chunk.toString());
      
      // Check if we have too many pending requests
      if (this.conversionState.pendingRequests.size >= this.conversionState.maxPending) {
        callback(new Error('Too many pending requests'));
        return;
      }
      
      // Map to target protocol format
      const convertedRequest = this._convertRequestFormat(request);
      const requestId = convertedRequest.id;
      
      // Save original request for later matching with response
      this.conversionState.pendingRequests.set(requestId, request);
      
      // Send to target protocol
      this.targetProtocol.sendRequest(convertedRequest, (err) => {
        if (err) {
          this.conversionState.pendingRequests.delete(requestId);
          callback(err);
          return;
        }
        callback();
      });
    } catch (err) {
      callback(new Error(`Protocol conversion error: ${err.message}`));
    }
  }
  
  // Protocol conversion methods
  _convertRequestFormat(sourceRequest) {
    // Implementation would convert between protocol formats
    return {
      id: sourceRequest.requestId || Date.now(),
      method: sourceRequest.action,
      params: sourceRequest.data,
      target: sourceRequest.endpoint
    };
  }
  
  _convertResponseFormat(targetResponse, originalRequest) {
    // Implementation would convert back to source protocol format
    return JSON.stringify({
      requestId: originalRequest.requestId,
      status: targetResponse.success ? 'success' : 'error',
      data: targetResponse.result,
      metadata: {
        timestamp: Date.now(),
        originalSource: originalRequest.source
      }
    });
  }
}
        

4. Transform Streams

Transform streams extend Duplex streams but with a unified interface where the output is a transformed version of the input.

Key implementation aspects:

• Implementing the _transform(chunk, encoding, callback) method that processes and transforms data
• Optional _flush(callback) method for handling end-of-stream operations
• State management for partial chunks and transformation context

const { Transform } = require('stream');
const crypto = require('crypto');

class BlockCipher extends Transform {
  constructor(options = {}) {
    super(options);
    // Cryptographic parameters
    this.algorithm = options.algorithm || 'aes-256-ctr';
    this.key = options.key || crypto.randomBytes(32);
    this.iv = options.iv || crypto.randomBytes(16);
    this.mode = options.mode || 'encrypt';
    
    // Block handling state
    this.blockSize = options.blockSize || 16;
    this.partialBlock = Buffer.alloc(0);
    
    // Create cipher based on mode
    this.cipher = this.mode === 'encrypt' 
      ? crypto.createCipheriv(this.algorithm, this.key, this.iv)
      : crypto.createDecipheriv(this.algorithm, this.key, this.iv);
      
    // Optional parameters
    this.autopadding = options.autopadding !== undefined ? options.autopadding : true;
    this.cipher.setAutoPadding(this.autopadding);
  }

  _transform(chunk, encoding, callback) {
    try {
      // Combine with any partial block from previous chunks
      const data = Buffer.concat([this.partialBlock, chunk]);
      
      // Process complete blocks
      const blocksToProcess = Math.floor(data.length / this.blockSize);
      const bytesToProcess = blocksToProcess * this.blockSize;
      
      if (bytesToProcess > 0) {
        // Process complete blocks
        const completeBlocks = data.slice(0, bytesToProcess);
        const transformedData = this.cipher.update(completeBlocks);
        
        // Save remaining partial block for next _transform call
        this.partialBlock = data.slice(bytesToProcess);
        
        // Push transformed data
        this.push(transformedData);
      } else {
        // Not enough data for even one block
        this.partialBlock = data;
      }
      
      callback();
    } catch (err) {
      callback(new Error(`Encryption error: ${err.message}`));
    }
  }

  _flush(callback) {
    try {
      // Process any remaining partial block
      let finalBlock = Buffer.alloc(0);
      
      if (this.partialBlock.length > 0) {
        finalBlock = this.cipher.update(this.partialBlock);
      }
      
      // Get final block from cipher
      const finalOutput = Buffer.concat([
        finalBlock,
        this.cipher.final()
      ]);
      
      // Push final data
      if (finalOutput.length > 0) {
        this.push(finalOutput);
      }
      
      // Add encryption metadata if in encryption mode
      if (this.mode === 'encrypt') {
        // Push metadata as JSON at end of stream
        this.push(JSON.stringify({
          algorithm: this.algorithm,
          iv: this.iv.toString('hex'),
          keyId: this._getKeyId(), // Reference to key rather than key itself
          format: 'hex'
        }));
      }
      
      callback();
    } catch (err) {
      callback(new Error(`Finalization error: ${err.message}`));
    }
  }
  
  _getKeyId() {
    // In a real implementation, this would return a key identifier
    // rather than the actual key
    return crypto.createHash('sha256').update(this.key).digest('hex').substring(0, 8);
  }
}
        

Internal Structure (Simplified):

// Simplified version of how EventEmitter works internally
class EventEmitter {
  constructor() {
    this._events = {}; // Internal registry of events and listeners
    this._maxListeners = 10; // Default limit before warning
  }
  
  // Add listener for an event
  on(eventName, listener) {
    if (!this._events[eventName]) {
      this._events[eventName] = [];
    }
    
    this._events[eventName].push(listener);
    
    // Check if we have too many listeners
    if (this._events[eventName].length > this._maxListeners) {
      console.warn(`Possible memory leak: ${this._events[eventName].length} 
                    listeners added for ${eventName}`);
    }
    
    return this;
  }
  
  // Emit event with arguments
  emit(eventName, ...args) {
    if (!this._events[eventName]) return false;
    
    const listeners = this._events[eventName].slice(); // Create a copy to avoid mutation issues
    
    for (const listener of listeners) {
      listener.apply(this, args);
    }
    
    return true;
  }
  
  // Other methods like once(), removeListener(), etc.
}
        

Advanced Usage with Error Handling:

const EventEmitter = require('events');
const fs = require('fs');

class FileProcessor extends EventEmitter {
  constructor(filePath) {
    super();
    this.filePath = filePath;
    this.data = null;
    
    // Best practice: Always have an error handler
    this.on('error', (err) => {
      console.error('Error in FileProcessor:', err);
      // Prevent uncaught exceptions
    });
  }
  
  processFile() {
    fs.readFile(this.filePath, 'utf8', (err, data) => {
      if (err) {
        this.emit('error', err);
        return;
      }
      
      try {
        this.data = JSON.parse(data);
        this.emit('processed', this.data);
      } catch (err) {
        this.emit('error', new Error(`Invalid JSON in file: ${err.message}`));
      }
    });
    
    return this; // Allow chaining
  }
}

// Usage
const processor = new FileProcessor('./config.json')
  .on('processed', (data) => {
    console.log('Config loaded:', data);
  })
  .processFile();
        

Optimizing for Performance:

class BufferedEventEmitter extends EventEmitter {
  constructor(options = {}) {
    super();
    this.buffers = new Map();
    this.flushInterval = options.flushInterval || 1000;
    this.maxBufferSize = options.maxBufferSize || 1000;
    this.timers = new Map();
  }
  
  bufferEvent(event, data) {
    if (!this.buffers.has(event)) {
      this.buffers.set(event, []);
    }
    
    const buffer = this.buffers.get(event);
    buffer.push(data);
    
    // Flush if we reach max buffer size
    if (buffer.length >= this.maxBufferSize) {
      this.flushEvent(event);
      return;
    }
    
    // Set up timed flush if not already scheduled
    if (!this.timers.has(event)) {
      const timerId = setTimeout(() => {
        this.flushEvent(event);
      }, this.flushInterval);
      
      this.timers.set(event, timerId);
    }
  }
  
  flushEvent(event) {
    if (this.timers.has(event)) {
      clearTimeout(this.timers.get(event));
      this.timers.delete(event);
    }
    
    if (!this.buffers.has(event) || this.buffers.get(event).length === 0) {
      return;
    }
    
    const items = this.buffers.get(event);
    this.buffers.set(event, []);
    
    // Emit the buffered batch
    super.emit(`${event}:batch`, items);
  }
  
  // Clean up all timers
  destroy() {
    for (const timerId of this.timers.values()) {
      clearTimeout(timerId);
    }
    this.timers.clear();
    this.buffers.clear();
    this.removeAllListeners();
  }
}

// Usage example for high-frequency events
const metrics = new BufferedEventEmitter({ 
  flushInterval: 5000,
  maxBufferSize: 500 
});

// Set up batch listener
metrics.on('dataPoint:batch', (dataPoints) => {
  console.log(`Processing ${dataPoints.length} data points in batch`);
  // Process in bulk - much more efficient
  db.bulkInsert(dataPoints);
});

// In high-frequency code
function recordMetric(value) {
  metrics.bufferEvent('dataPoint', {
    value,
    timestamp: Date.now()
  });
}
        

Buffer Creation Methods:

// Allocate a new buffer (initialized with zeros)
const buffer1 = Buffer.alloc(10);  // Creates a zero-filled Buffer of length 10

// Allocate uninitialized buffer (faster but contains old memory data)
const buffer2 = Buffer.allocUnsafe(10);  // Faster allocation, but may contain sensitive data

// Create from existing data
const buffer3 = Buffer.from([0x62, 0x75, 0x66, 0x66, 0x65, 0x72]);  // From array of bytes
const buffer4 = Buffer.from('buffer', 'utf8');  // From string with encoding
        

Creation Methods with Performance Considerations:

// Safe allocation (zeroed memory)
// Performance: Slightly slower due to zero-filling
// Use when: Security is important or when you need a clean buffer
const safeBuffer = Buffer.alloc(1024);

// Unsafe allocation (faster but may contain old data)
// Performance: Faster allocation, no initialization overhead
// Use when: Performance is critical and you will immediately overwrite the entire buffer
const fastBuffer = Buffer.allocUnsafe(1024);

// Pre-filled allocation
// Performance: Similar to alloc() but saves a step when you need a specific fill value
// Use when: You need a buffer initialized with a specific byte value
const filledBuffer = Buffer.alloc(1024, 0xFF);  // All bytes set to 255

// From existing data
// Performance: Depends on input type; typed arrays are fastest
// Use when: Converting between data formats
const fromStringBuffer = Buffer.from('binary data', 'utf8');
const fromArrayBuffer = Buffer.from(new Uint8Array([1, 2, 3]));  // Zero-copy for TypedArrays
const fromBase64 = Buffer.from('SGVsbG8gV29ybGQ=', 'base64');
        

Efficient Buffer Operations:

// In-place manipulation (better performance, no additional allocations)
function inPlaceTransform(buffer) {
    for (let i = 0; i < buffer.length; i++) {
        buffer[i] = buffer[i] ^ 0xFF;  // Bitwise XOR (toggles all bits)
    }
    return buffer;  // Original buffer is modified
}

// Buffer pooling for frequent small allocations
function efficientProcessing() {
    // Reuse the same buffer for multiple operations to reduce GC pressure
    const reuseBuffer = Buffer.allocUnsafe(1024);
    
    for (let i = o; i < 1000; i++) {
        // Use the same buffer for each operation
        // Fill with new data each time
        reuseBuffer.fill(0);  // Reset the buffer
        // Process data using reuseBuffer...
    }
}

// Working with binary structures
function readInt32BE(buffer, offset = 0) {
    return buffer.readInt32BE(offset);
}

function writeStruct(buffer, value, position) {
    // Write a complex structure to a buffer at a specific position
    let offset = position;
    
    // Write 32-bit integer in big-endian format
    offset = buffer.writeUInt32BE(value.id, offset);
    
    // Write 16-bit integer in little-endian format
    offset = buffer.writeUInt16LE(value.flags, offset);
    
    // Write a fixed-length string
    offset += buffer.write(value.name.padEnd(16, '\\0'), offset, 16);
    
    return offset;  // Return new position after write
}
        

Buffer Transformations and Performance Optimization:

// Buffer slicing (zero-copy view)
const buffer = Buffer.from('Hello World');
const view = buffer.slice(0, 5);  // Creates a view, shares underlying memory

// IMPORTANT: slice() creates a view - modifications affect the original buffer
view[0] = 74;  // ASCII for 'J'
console.log(buffer.toString());  // Outputs: "Jello World"

// To create a real copy instead of a view:
const copy = Buffer.allocUnsafe(5);
buffer.copy(copy, 0, 0, 5);
copy[0] = 77;  // ASCII for 'M'
console.log(buffer.toString());  // Still: "Jello World" (original unchanged)

// Efficient concatenation with pre-allocation
function optimizedConcat(buffers) {
    // Calculate total length first to avoid multiple allocations
    const totalLength = buffers.reduce((acc, buf) => acc + buf.length, 0);
    
    // Pre-allocate the final buffer once
    const result = Buffer.allocUnsafe(totalLength);
    
    let offset = 0;
    for (const buf of buffers) {
        buf.copy(result, offset);
        offset += buf.length;
    }
    
    return result;
}

// Buffer comparison (constant time for security-sensitive applications)
function constantTimeCompare(bufA, bufB) {
    if (bufA.length !== bufB.length) return false;
    
    let diff = 0;
    for (let i = 0; i < bufA.length; i++) {
        // XOR will be 0 for matching bytes, non-zero for different bytes
        diff |= bufA[i] ^ bufB[i];
    }
    
    return diff === 0;
}
        

Working with Different Encodings:

const buffer = Buffer.from('Hello World');

// Convert to different string encodings
const hex = buffer.toString('hex');  // 48656c6c6f20576f726c64
const base64 = buffer.toString('base64');  // SGVsbG8gV29ybGQ=
const binary = buffer.toString('binary');  // Binary encoding

// Handling multi-byte characters in UTF-8
const utf8Buffer = Buffer.from('🔥火🔥', 'utf8');
console.log(utf8Buffer.length);  // 10 bytes (not 3 characters)
console.log(utf8Buffer);  // 

// Detecting incomplete UTF-8 sequences
function isCompleteUtf8(buffer) {
    // Check the last few bytes to see if we have an incomplete multi-byte sequence
    if (buffer.length === 0) return true;
    
    const lastByte = buffer[buffer.length - 1];
    
    // If the last byte is a continuation byte (10xxxxxx) or start of multi-byte sequence
    if ((lastByte & 0x80) === 0) return true;  // ASCII byte
    if ((lastByte & 0xC0) === 0x80) return false;  // Continuation byte
    
    if ((lastByte & 0xE0) === 0xC0) return buffer.length >= 2;  // 2-byte sequence
    if ((lastByte & 0xF0) === 0xE0) return buffer.length >= 3;  // 3-byte sequence
    if ((lastByte & 0xF8) === 0xF0) return buffer.length >= 4;  // 4-byte sequence
    
    return false;  // Invalid UTF-8 start byte
}
        

Advanced Implementation with Thread Pool:

const { Worker } = require('worker_threads');
const os = require('os');

class ThreadPool {
  constructor(size = os.cpus().length) {
    this.size = size;
    this.workers = [];
    this.queue = [];
    this.activeWorkers = 0;
    
    // Initialize worker pool
    for (let i = 0; i < this.size; i++) {
      this.workers.push({
        worker: null,
        isWorking: false,
        id: i
      });
    }
  }

  runTask(workerScript, workerData) {
    return new Promise((resolve, reject) => {
      const task = { workerScript, workerData, resolve, reject };
      
      // Try to run task immediately or queue it
      const availableWorker = this.workers.find(w => !w.isWorking);
      if (availableWorker) {
        this._runWorker(availableWorker, task);
      } else {
        this.queue.push(task);
      }
    });
  }

  _runWorker(workerObj, task) {
    workerObj.isWorking = true;
    this.activeWorkers++;
    
    // Create new worker with the provided script
    workerObj.worker = new Worker(task.workerScript, { 
      workerData: task.workerData 
    });
    
    // Handle messages
    workerObj.worker.on('message', (result) => {
      task.resolve(result);
      this._cleanupWorker(workerObj);
    });
    
    // Handle errors
    workerObj.worker.on('error', (err) => {
      task.reject(err);
      this._cleanupWorker(workerObj);
    });
    
    // Handle worker exit
    workerObj.worker.on('exit', (code) => {
      if (code !== 0) {
        task.reject(new Error(`Worker stopped with exit code ${code}`));
      }
      this._cleanupWorker(workerObj);
    });
  }

  _cleanupWorker(workerObj) {
    workerObj.isWorking = false;
    workerObj.worker = null;
    this.activeWorkers--;
    
    // Process queue if there are pending tasks
    if (this.queue.length > 0) {
      const nextTask = this.queue.shift();
      this._runWorker(workerObj, nextTask);
    }
  }

  getActiveCount() {
    return this.activeWorkers;
  }

  getQueueLength() {
    return this.queue.length;
  }
}

// Usage
const pool = new ThreadPool();
const promises = [];

// Add 20 tasks to our thread pool
for (let i = 0; i < 20; i++) {
  promises.push(pool.runTask('./worker-script.js', { taskId: i }));
}

Promise.all(promises).then(results => {
  console.log('All tasks completed', results);
});
        

Performance Comparison of Data Sharing Methods:

// Transferring a large buffer (faster, zero-copy)
const buffer = new ArrayBuffer(100 * 1024 * 1024); // 100MB buffer
worker.postMessage({ buffer }, [buffer]); // Second arg is transfer list

// Using SharedArrayBuffer (best for frequent updates)
const sharedBuffer = new SharedArrayBuffer(100 * 1024 * 1024);
const uint8 = new Uint8Array(sharedBuffer);
// Write to buffer
uint8[0] = 1;
// Both threads can now read/write to this memory
worker.postMessage({ sharedBuffer });
        

Advanced spawn() Implementation with Error Handling and Timeout:

const { spawn } = require('child_process');
const fs = require('fs');

function executeCommand(command, args, options = {}) {
  return new Promise((resolve, reject) => {
    // Default options with sensible security values
    const defaultOptions = {
      cwd: process.cwd(),
      env: process.env,
      shell: false,
      timeout: 30000, // 30 seconds
      maxBuffer: 1024 * 1024, // 1MB
      ...options
    };
    
    // Create output streams if requested
    const stdout = options.outputFile ? 
      fs.createWriteStream(options.outputFile) : null;
    
    // Launch process
    const child = spawn(command, args, defaultOptions);
    let stdoutData = '';
    let stderrData = '';
    let killed = false;
    
    // Set timeout if specified
    const timeoutId = defaultOptions.timeout ? 
      setTimeout(() => {
        killed = true;
        child.kill('SIGTERM');
        setTimeout(() => {
          child.kill('SIGKILL');
        }, 2000); // Force kill after 2 seconds
        reject(new Error(`Command timed out after ${defaultOptions.timeout}ms: ${command}`));
      }, defaultOptions.timeout) : null;
    
    // Handle standard output
    child.stdout.on('data', (data) => {
      if (stdout) {
        stdout.write(data);
      }
      
      // Only store data if we're not streaming to a file
      if (!stdout && stdoutData.length < defaultOptions.maxBuffer) {
        stdoutData += data;
      } else if (!stdout && stdoutData.length >= defaultOptions.maxBuffer) {
        killed = true;
        child.kill('SIGTERM');
        reject(new Error(`Maximum buffer size exceeded for stdout: ${command}`));
      }
    });
    
    // Handle standard error
    child.stderr.on('data', (data) => {
      if (stderrData.length < defaultOptions.maxBuffer) {
        stderrData += data;
      } else if (stderrData.length >= defaultOptions.maxBuffer) {
        killed = true;
        child.kill('SIGTERM');
        reject(new Error(`Maximum buffer size exceeded for stderr: ${command}`));
      }
    });
    
    // Handle process close
    child.on('close', (code) => {
      if (timeoutId) clearTimeout(timeoutId);
      if (stdout) stdout.end();
      
      if (!killed) {
        resolve({
          code,
          stdout: stdoutData,
          stderr: stderrData
        });
      }
    });
    
    // Handle process errors
    child.on('error', (error) => {
      if (timeoutId) clearTimeout(timeoutId);
      reject(new Error(`Failed to start process ${command}: ${error.message}`));
    });
  });
}

// Example usage with pipe to file
executeCommand('ffmpeg', ['-i', 'input.mp4', 'output.mp4'], {
  outputFile: 'transcoding.log',
  timeout: 60000 // 1 minute
})
.then(result => console.log('Process completed with code:', result.code))
.catch(err => console.error('Process failed:', err));
        

Implementing a Secure exec() Wrapper with Input Sanitization:

const { exec } = require('child_process');
const childProcess = require('child_process');
const util = require('util');

// Promisify exec for cleaner async/await usage
const execPromise = util.promisify(childProcess.exec);

// Safe command execution that prevents command injection
async function safeExec(command, args = [], options = {}) {
  // Validate input command
  if (typeof command !== 'string' || !command.trim()) {
    throw new Error('Invalid command specified');
  }
  
  // Validate and sanitize arguments
  if (!Array.isArray(args)) {
    throw new Error('Arguments must be an array');
  }
  
  // Properly escape arguments to prevent injection
  const escapedArgs = args.map(arg => {
    // Convert to string and escape special characters
    const str = String(arg);
    // Different escaping for Windows vs Unix
    if (process.platform === 'win32') {
      // Windows escaping: double quotes and escape inner quotes
      return `"${str.replace(/"/g, '""')}"`;
    } else {
      // Unix escaping with single quotes
      return `'${str.replace(/\'/g, '\\'\')'`;
    }
  });
  
  // Construct safe command string
  const safeCommand = `${command} ${escapedArgs.join(' ')}`;
  
  try {
    // Execute with timeout and maxBuffer settings
    const defaultOptions = {
      timeout: 30000,
      maxBuffer: 1024 * 1024,
      ...options
    };
    
    const { stdout, stderr } = await execPromise(safeCommand, defaultOptions);
    return { stdout, stderr, exitCode: 0 };
  } catch (error) {
    // Handle exec errors (non-zero exit code, timeout, etc.)
    return {
      stdout: error.stdout || '',
      stderr: error.stderr || error.message,
      exitCode: error.code || 1,
      error
    };
  }
}

// Example usage
async function main() {
  // Safe way to execute a command with user input
  const userInput = process.argv[2] || 'text file.txt';
  
  try {
    // Instead of dangerously doing: exec(`grep ${userInput} *`)
    const result = await safeExec('grep', [userInput, '*']);
    
    if (result.exitCode === 0) {
      console.log('Command output:', result.stdout);
    } else {
      console.error('Command failed:', result.stderr);
    }
  } catch (err) {
    console.error('Execution error:', err);
  }
}

main();
        

execFile() with Environment Control and Process Priority:

const { execFile } = require('child_process');
const path = require('path');
const os = require('os');

function runExecutable(executablePath, args, options = {}) {
  return new Promise((resolve, reject) => {
    // Normalize path for cross-platform compatibility
    const normalizedPath = path.normalize(executablePath);
    
    // Create isolated environment with specific variables
    const customEnv = {
      // Start with clean slate or inherited environment
      ...(options.cleanEnv ? {} : process.env),
      // Add custom environment variables
      ...(options.env || {}),
      // Set specific Node.js runtime settings
      NODE_OPTIONS: options.nodeOptions || process.env.NODE_OPTIONS || ''
    };
    
    // Platform-specific settings for process priority
    let platformOptions = {};
    
    if (process.platform === 'win32' && options.priority) {
      // Windows process priority
      platformOptions.windowsHide = true;
      
      // Map priority names to Windows priority classes
      const priorityMap = {
        low: 0x00000040,      // IDLE_PRIORITY_CLASS
        belowNormal: 0x00004000, // BELOW_NORMAL_PRIORITY_CLASS
        normal: 0x00000020,    // NORMAL_PRIORITY_CLASS
        aboveNormal: 0x00008000, // ABOVE_NORMAL_PRIORITY_CLASS
        high: 0x00000080,     // HIGH_PRIORITY_CLASS
        realtime: 0x00000100  // REALTIME_PRIORITY_CLASS (use with caution)
      };
      
      if (priorityMap[options.priority]) {
        platformOptions.windowsPriority = priorityMap[options.priority];
      }
    } else if ((process.platform === 'linux' || process.platform === 'darwin') && options.priority) {
      // For Unix systems, we'll prefix with nice command in the wrapper
      // This is handled separately below
    }
    
    // Configure execution options
    const execOptions = {
      env: customEnv,
      timeout: options.timeout || 0,
      maxBuffer: options.maxBuffer || 1024 * 1024 * 10, // 10MB
      killSignal: options.killSignal || 'SIGTERM',
      cwd: options.cwd || process.cwd(),
      ...platformOptions
    };
    
    // Handle Linux/macOS nice level by using a wrapper if needed
    if ((process.platform === 'linux' || process.platform === 'darwin') && options.priority) {
      const niceMap = {
        realtime: -20, // Requires root
        high: -10,
        aboveNormal: -5,
        normal: 0,
        belowNormal: 5,
        low: 10
      };
      
      const niceLevel = niceMap[options.priority] || 0;
      
      // If nice level requires root but we're not root, fall back to normal execution
      if (niceLevel < 0 && os.userInfo().uid !== 0) {
        console.warn(`Warning: Requested priority ${options.priority} requires root privileges. Using normal priority.`);
        // Proceed with normal execFile below
      } else {
        // Use nice with specified level
        return new Promise((resolve, reject) => {
          execFile('nice', [`-n${niceLevel}`, normalizedPath, ...args], execOptions, 
            (error, stdout, stderr) => {
              if (error) {
                reject(error);
              } else {
                resolve({ stdout, stderr });
              }
            });
        });
      }
    }
    
    // Standard execFile execution
    execFile(normalizedPath, args, execOptions, (error, stdout, stderr) => {
      if (error) {
        reject(error);
      } else {
        resolve({ stdout, stderr });
      }
    });
  });
}

// Example usage
async function processImage() {
  try {
    // Run an image processing tool with high priority
    const result = await runExecutable('convert', 
      ['input.jpg', '-resize', '50%', 'output.jpg'], 
      {
        priority: 'high',
        env: { MAGICK_THREAD_LIMIT: '4' }, // Control ImageMagick threads
        timeout: 60000 // 1 minute timeout
      }
    );
    
    console.log('Image processing complete');
    return result;
  } catch (error) {
    console.error('Image processing failed:', error);
    throw error;
  }
}
        

Worker Pool Implementation with fork():

// main.js - Worker Pool Manager
const { fork } = require('child_process');
const os = require('os');
const EventEmitter = require('events');

class NodeWorkerPool extends EventEmitter {
  constructor(workerScript, options = {}) {
    super();
    this.workerScript = workerScript;
    this.options = {
      maxWorkers: options.maxWorkers || os.cpus().length,
      minWorkers: options.minWorkers || 1,
      maxTasksPerWorker: options.maxTasksPerWorker || 10,
      idleTimeout: options.idleTimeout || 30000, // 30 seconds
      taskTimeout: options.taskTimeout || 60000, // 1 minute
      ...options
    };
    
    this.workers = [];
    this.taskQueue = [];
    this.workersById = new Map();
    this.workerStatus = new Map();
    this.tasksByWorkerId = new Map();
    this.idleTimers = new Map();
    this.taskTimeouts = new Map();
    this.taskCounter = 0;
    
    // Initialize minimum number of workers
    this._initializeWorkers();
    
    // Start monitoring system load for auto-scaling
    if (this.options.autoScale) {
      this._startLoadMonitoring();
    }
  }
  
  _initializeWorkers() {
    for (let i = 0; i < this.options.minWorkers; i++) {
      this._createWorker();
    }
  }
  
  _createWorker() {
    const worker = fork(this.workerScript, [], {
      env: { ...process.env, ...this.options.env },
      execArgv: this.options.execArgv || []
    });
    
    const workerId = worker.pid;
    this.workers.push(worker);
    this.workersById.set(workerId, worker);
    this.workerStatus.set(workerId, { status: 'idle', tasksCompleted: 0 });
    this.tasksByWorkerId.set(workerId, new Set());
    
    // Set up message handling
    worker.on('message', (message) => {
      if (message.type === 'task:completed') {
        this._handleTaskCompletion(workerId, message);
      } else if (message.type === 'worker:ready') {
        this._assignTaskIfAvailable(workerId);
      } else if (message.type === 'worker:error') {
        this._handleWorkerError(workerId, message.error);
      }
    });
    
    // Handle worker exit
    worker.on('exit', (code, signal) => {
      this._handleWorkerExit(workerId, code, signal);
    });
    
    // Handle errors
    worker.on('error', (error) => {
      this._handleWorkerError(workerId, error);
    });
    
    // Start idle timer
    this._resetIdleTimer(workerId);
    
    return workerId;
  }
  
  _resetIdleTimer(workerId) {
    // Clear existing timer
    if (this.idleTimers.has(workerId)) {
      clearTimeout(this.idleTimers.get(workerId));
    }
    
    // Set new timer only if we have more than minimum workers
    if (this.workers.length > this.options.minWorkers) {
      this.idleTimers.set(workerId, setTimeout(() => {
        // If worker is idle and we have more than minimum workers, terminate it
        if (this.workerStatus.get(workerId).status === 'idle') {
          this._terminateWorker(workerId);
        }
      }, this.options.idleTimeout));
    }
  }
  
  _assignTaskIfAvailable(workerId) {
    if (this.taskQueue.length > 0) {
      const task = this.taskQueue.shift();
      this._assignTaskToWorker(workerId, task);
    } else {
      this.workerStatus.set(workerId, { 
        ...this.workerStatus.get(workerId), 
        status: 'idle' 
      });
      this._resetIdleTimer(workerId);
    }
  }
  
  _assignTaskToWorker(workerId, task) {
    const worker = this.workersById.get(workerId);
    if (!worker) return false;
    
    this.workerStatus.set(workerId, { 
      ...this.workerStatus.get(workerId), 
      status: 'busy' 
    });
    
    // Clear idle timer
    if (this.idleTimers.has(workerId)) {
      clearTimeout(this.idleTimers.get(workerId));
      this.idleTimers.delete(workerId);
    }
    
    // Set task timeout
    this.taskTimeouts.set(task.id, setTimeout(() => {
      this._handleTaskTimeout(task.id, workerId);
    }, this.options.taskTimeout));
    
    // Track this task
    this.tasksByWorkerId.get(workerId).add(task.id);
    
    // Send task to worker
    worker.send({
      type: 'task:execute',
      taskId: task.id,
      payload: task.payload
    });
    
    return true;
  }
  
  _handleTaskCompletion(workerId, message) {
    const taskId = message.taskId;
    const result = message.result;
    const error = message.error;
    
    // Clear task timeout
    if (this.taskTimeouts.has(taskId)) {
      clearTimeout(this.taskTimeouts.get(taskId));
      this.taskTimeouts.delete(taskId);
    }
    
    // Update worker stats
    if (this.workerStatus.has(workerId)) {
      const status = this.workerStatus.get(workerId);
      this.workerStatus.set(workerId, {
        ...status,
        tasksCompleted: status.tasksCompleted + 1
      });
    }
    
    // Remove task from tracking
    this.tasksByWorkerId.get(workerId).delete(taskId);
    
    // Resolve or reject the task promise
    const taskPromise = this.taskPromises.get(taskId);
    if (taskPromise) {
      if (error) {
        taskPromise.reject(new Error(error));
      } else {
        taskPromise.resolve(result);
      }
      this.taskPromises.delete(taskId);
    }
    
    // Check if worker should be recycled based on tasks completed
    const tasksCompleted = this.workerStatus.get(workerId).tasksCompleted;
    if (tasksCompleted >= this.options.maxTasksPerWorker) {
      this._recycleWorker(workerId);
    } else {
      // Assign next task or mark as idle
      this._assignTaskIfAvailable(workerId);
    }
  }
  
  _handleTaskTimeout(taskId, workerId) {
    const worker = this.workersById.get(workerId);
    const taskPromise = this.taskPromises.get(taskId);
    
    // Reject the task promise
    if (taskPromise) {
      taskPromise.reject(new Error(`Task ${taskId} timed out after ${this.options.taskTimeout}ms`));
      this.taskPromises.delete(taskId);
    }
    
    // Recycle the worker as it might be stuck
    this._recycleWorker(workerId);
  }
  
  // Public API to execute a task
  executeTask(payload) {
    this.taskCounter++;
    const taskId = `task-${Date.now()}-${this.taskCounter}`;
    
    // Create a promise for this task
    const taskPromise = {};
    const promise = new Promise((resolve, reject) => {
      taskPromise.resolve = resolve;
      taskPromise.reject = reject;
    });
    this.taskPromises = this.taskPromises || new Map();
    this.taskPromises.set(taskId, taskPromise);
    
    // Create the task object
    const task = {
      id: taskId,
      payload,
      addedAt: Date.now()
    };
    
    // Find an idle worker or queue the task
    const idleWorker = Array.from(this.workerStatus.entries())
      .find(([id, status]) => status.status === 'idle');
    
    if (idleWorker) {
      this._assignTaskToWorker(idleWorker[0], task);
    } else if (this.workers.length < this.options.maxWorkers) {
      // Create a new worker if we haven't reached the limit
      const newWorkerId = this._createWorker();
      this._assignTaskToWorker(newWorkerId, task);
    } else {
      // Queue the task for later execution
      this.taskQueue.push(task);
    }
    
    return promise;
  }
  
  // Helper methods for worker lifecycle management
  _recycleWorker(workerId) {
    // Create a replacement worker
    this._createWorker();
    
    // Gracefully terminate the old worker
    this._terminateWorker(workerId);
  }
  
  _terminateWorker(workerId) {
    const worker = this.workersById.get(workerId);
    if (!worker) return;
    
    // Clean up all resources
    if (this.idleTimers.has(workerId)) {
      clearTimeout(this.idleTimers.get(workerId));
      this.idleTimers.delete(workerId);
    }
    
    // Reassign any pending tasks
    const pendingTasks = this.tasksByWorkerId.get(workerId);
    if (pendingTasks && pendingTasks.size > 0) {
      for (const taskId of pendingTasks) {
        // Add back to queue with high priority
        const taskPromise = this.taskPromises.get(taskId);
        if (taskPromise) {
          this.taskQueue.unshift({
            id: taskId,
            payload: { retryFromWorker: workerId }
          });
        }
      }
    }
    
    // Remove from tracking
    this.workersById.delete(workerId);
    this.workerStatus.delete(workerId);
    this.tasksByWorkerId.delete(workerId);
    this.workers = this.workers.filter(w => w.pid !== workerId);
    
    // Send graceful termination signal
    worker.send({ type: 'worker:shutdown' });
    
    // Force kill after timeout
    setTimeout(() => {
      if (!worker.killed) {
        worker.kill('SIGKILL');
      }
    }, 5000);
  }
  
  // Shut down the pool
  shutdown() {
    // Stop accepting new tasks
    this.shuttingDown = true;
    
    // Wait for all tasks to complete or timeout
    return new Promise((resolve) => {
      const pendingTasks = this.taskPromises ? this.taskPromises.size : 0;
      
      if (pendingTasks === 0) {
        this._forceShutdown();
        resolve();
      } else {
        console.log(`Waiting for ${pendingTasks} tasks to complete...`);
        
        // Set a maximum wait time
        const shutdownTimeout = setTimeout(() => {
          console.log('Shutdown timeout reached, forcing termination');
          this._forceShutdown();
          resolve();
        }, 30000); // 30 seconds max wait
        
        // Check periodically if all tasks are done
        const checkInterval = setInterval(() => {
          const remainingTasks = this.taskPromises ? this.taskPromises.size : 0;
          if (remainingTasks === 0) {
            clearInterval(checkInterval);
            clearTimeout(shutdownTimeout);
            this._forceShutdown();
            resolve();
          }
        }, 500);
      }
    });
  }
  
  _forceShutdown() {
    // Terminate all workers
    for (const worker of this.workers) {
      worker.removeAllListeners();
      if (!worker.killed) {
        worker.kill('SIGTERM');
      }
    }
    
    // Clear all timers
    for (const timerId of this.idleTimers.values()) {
      clearTimeout(timerId);
    }
    for (const timerId of this.taskTimeouts.values()) {
      clearTimeout(timerId);
    }
    
    // Clear all tracking data
    this.workers = [];
    this.workersById.clear();
    this.workerStatus.clear();
    this.tasksByWorkerId.clear();
    this.idleTimers.clear();
    this.taskTimeouts.clear();
    this.taskQueue = [];
    
    if (this.loadMonitorInterval) {
      clearInterval(this.loadMonitorInterval);
    }
  }
  
  // Auto-scaling based on system load
  _startLoadMonitoring() {
    this.loadMonitorInterval = setInterval(() => {
      const currentLoad = os.loadavg()[0] / os.cpus().length; // Normalized load
      
      if (currentLoad > 0.8 && this.workers.length < this.options.maxWorkers) {
        // System is heavily loaded, add workers
        this._createWorker();
      } else if (currentLoad < 0.2 && this.workers.length > this.options.minWorkers) {
        // System is lightly loaded, can reduce workers (idle ones will timeout)
        // We don't actively reduce here, idle timeouts will handle it
      }
    }, 30000); // Check every 30 seconds
  }
}

// Example worker.js implementation
/*
process.on('message', (message) => {
  if (message.type === 'task:execute') {
    // Process the task
    try {
      // Do some work based on message.payload
      const result = someFunction(message.payload);
      
      // Send result back
      process.send({
        type: 'task:completed',
        taskId: message.taskId,
        result
      });
    } catch (error) {
      process.send({
        type: 'task:completed',
        taskId: message.taskId,
        error: error.message
      });
    }
  } else if (message.type === 'worker:shutdown') {
    // Clean up and exit gracefully
    process.exit(0);
  }
});

// Signal that we're ready to process tasks
process.send({ type: 'worker:ready' });
*/

// Example usage
const pool = new NodeWorkerPool('./worker.js', {
  minWorkers: 2,
  maxWorkers: 8,
  autoScale: true
});

// Execute some tasks
async function runTasks() {
  const results = await Promise.all([
    pool.executeTask({ type: 'calculation', data: { x: 10, y: 20 } }),
    pool.executeTask({ type: 'processing', data: 'some text' }),
    // More tasks...
  ]);
  
  console.log('All tasks completed:', results);
  
  // Shut down the pool when done
  await pool.shutdown();
}

runTasks().catch(console.error);