// Simplified EventEmitter internals — the actual Node.js source is similar
class EventEmitter {
  private _events: Map<string, Function[]> = new Map()
  private _maxListeners: number = 10

  on(event: string, listener: Function): this {
    const existing = this._events.get(event) ?? []

    // ⚠️ MaxListeners warning: protect against accidental accumulation
    if (existing.length >= this._maxListeners) {
      process.emitWarning(
        `Possible EventEmitter memory leak detected. ${existing.length + 1} listeners ` +
        `added to event '${event}'. Use setMaxListeners(n) to increase limit.`
      )
    }

    this._events.set(event, [...existing, listener])
    return this
  }

  once(event: string, listener: Function): this {
    // Wrap listener to auto-remove itself after first call
    const wrapper = (...args: unknown[]) => {
      this.off(event, wrapper)
      listener.apply(this, args)
    }
    return this.on(event, wrapper)
  }

  off(event: string, listener: Function): this {
    const listeners = this._events.get(event)
    if (listeners) {
      this._events.set(event, listeners.filter(l => l !== listener))
    }
    return this
  }

  emit(event: string, ...args: unknown[]): boolean {
    const listeners = this._events.get(event)
    if (!listeners?.length) return false
    listeners.forEach(listener => listener.apply(this, args))
    return true
  }
}

// ❌ Listener accumulates on every request — classic Node.js memory leak
app.get('/stream', (req, res) => {
  const upstream = getUpstreamService()

  // ❌ A new listener is attached on EVERY request to /stream
  // upstream is a long-lived emitter — these listeners are NEVER removed
  upstream.on('data', (chunk) => res.write(chunk))
  upstream.on('end', () => res.end())
  // After 1000 requests: 1000 'data' listeners, 1000 'end' listeners on upstream
  // Node.js emits MaxListeners warning; memory grows linearly with request count
})

// ✅ Scoped listeners + cleanup
app.get('/stream', (req, res) => {
  const upstream = getUpstreamService()

  const onData = (chunk: Buffer) => res.write(chunk)
  const onEnd  = () => {
    res.end()
    upstream.off('data', onData)  // Remove data listener when stream ends
    // onEnd itself is attached with .once() — auto-removed after first call
  }

  upstream.on('data', onData)
  upstream.once('end', onEnd)

  // Also handle early client disconnect — remove listeners if client leaves
  req.on('close', () => {
    upstream.off('data', onData)
    upstream.off('end', onEnd)
  })
})

import { createReadStream, createWriteStream } from 'fs'
import { createGzip } from 'zlib'
import { pipeline } from 'stream/promises'
import { Transform } from 'stream'

// Custom Transform stream: uppercase each line of a CSV
class UppercaseTransform extends Transform {
  _transform(
    chunk: Buffer,
    encoding: BufferEncoding,
    callback: (error?: Error | null, data?: Buffer) => void
  ): void {
    // chunk is a Buffer; convert to string, transform, convert back
    const transformed = chunk.toString().toUpperCase()
    callback(null, Buffer.from(transformed))
    // callback() signals to the stream that we're ready for the next chunk
    // Without calling callback(), the stream stalls (backpressure applied)
  }
}

// ✅ Stream pipeline: entire 10 GB CSV processed with constant memory
await pipeline(
  createReadStream('huge.csv'),        // Readable: emits chunks (highWaterMark: 16KB default)
  new UppercaseTransform(),             // Transform: processes each chunk
  createGzip(),                         // Transform: gzip-compresses each chunk
  createWriteStream('huge.csv.gz')     // Writable: writes compressed chunks to disk
)
// At any point, only one chunk (16KB) is in memory between each stage
// The 10 GB file never fits entirely in memory

// ❌ The alternative — loads entire file into memory
const data = await readFile('huge.csv')  // 10 GB in memory
await writeFile('huge.csv.gz', gzipSync(data))
// Memory usage: 10 GB (original) + 10 GB (gzipped) = 20 GB peak

// Understanding backpressure manually (for custom stream implementations)
const readable = createReadStream('source.dat')
const writable = createWriteStream('dest.dat')

readable.on('data', (chunk) => {
  const canContinue = writable.write(chunk)
  // writable.write() returns:
  // true  → internal buffer has room → continue reading
  // false → internal buffer is full → STOP reading (backpressure signal)

  if (!canContinue) {
    readable.pause()  // ← Stop emitting data events
    writable.once('drain', () => {
      readable.resume()  // ← Buffer cleared → resume reading
    })
  }
})

readable.on('end', () => writable.end())

// ✅ This logic is exactly what stream.pipeline() handles for you automatically
// The above manual implementation is shown for understanding only
// Always use pipeline() in production code

// 64KB chunks instead of the default 16KB — useful for network streams where
// larger chunks amortize per-chunk overhead
const stream = createReadStream('file.bin', { highWaterMark: 64 * 1024 })

// worker.ts — runs in a separate V8 thread
import { parentPort, workerData } from 'worker_threads'
import { createHash } from 'crypto'

// workerData contains the data passed from the main thread
const { data, algorithm } = workerData as { data: string; algorithm: string }

// CPU-intensive work executes here — does NOT block main thread
const hash = createHash(algorithm).update(data).digest('hex')

// Send result back to main thread
parentPort?.postMessage({ hash })

// ===

// main.ts — orchestrates workers
import { Worker } from 'worker_threads'

function runInWorker<T>(
  workerPath: string,
  data: Record<string, unknown>
): Promise<T> {
  return new Promise((resolve, reject) => {
    const worker = new Worker(workerPath, { workerData: data })
    worker.on('message', resolve)
    worker.on('error', reject)
    worker.on('exit', (code) => {
      if (code !== 0) reject(new Error(`Worker exited with code ${code}`))
    })
  })
}

// Main thread: fires and forgets — event loop continues processing requests
const result = await runInWorker<{ hash: string }>('./worker.js', {
  data: 'large-payload-to-hash',
  algorithm: 'sha256',
})
console.log(result.hash)

// ✅ SharedArrayBuffer: zero-copy shared memory between main and worker
const sharedBuffer = new SharedArrayBuffer(4)  // 4 bytes = one Int32
const counter = new Int32Array(sharedBuffer)

// Pass buffer to worker (zero-copy — both threads access the same memory)
const worker = new Worker('./counter-worker.js', {
  workerData: { sharedBuffer }
})

// Main thread: read the counter value (written by worker)
Atomics.wait(counter, 0, 0)  // Wait until counter[0] !== 0
console.log('Worker incremented counter to:', counter[0])

// ===

// counter-worker.js
import { workerData } from 'worker_threads'
const counter = new Int32Array(workerData.sharedBuffer)

// Worker thread: atomically increment — safe across threads
Atomics.add(counter, 0, 1)
Atomics.notify(counter, 0, 1)  // Wake the main thread's Atomics.wait()