Processar uploads CSV grandes com Streams, Worker Threads e progresso via SSE em Node.js

1) Introducao ao problema

O que acontece quando você lê tudo na memória

Uma abordagem comum entre iniciantes é aceitar um upload, ler todo o arquivo para memória e depois parseá-lo de uma vez. Em Node.js isso parece simples: você usa multer com memoryStorage ou fs.readFile, transforma o conteúdo em string, passa para um parser síncrono e pronto. Funciona para arquivos pequenos, mas falha rapidamente quando o CSV cresce: memória estoura, GC trava a aplicação e o processo pode morrer com OOM.

Analogia simples: a esteira vs a caixa

Pense no processamento de CSV como uma fábrica. Ler o arquivo inteiro é como guardar todo o material em uma caixa gigante antes de processar — a caixa enche e bloqueia o fluxo. Usar streams é colocar uma esteira: cada linha passa, é processada e removida. Worker Threads são os operários que fazem o trabalho pesado sem travar o operador da esteira (o event loop). SSE é a placa de status que mostra o progresso em tempo real à equipe.

Antes de ver código correto, vou mostrar o jeito errado — porque é importante reconhecer padrões que quebram em produção. Depois do erro, mostro a correção com streams, workers e SSE.

2) Pre-requisitos e setup

Ferramentas e versões

Para acompanhar os exemplos você precisa de: Node.js 14+ (ideal 16+), npm/yarn, e conhecimento básico de JavaScript e linha de comando. Também usaremos o Express para endpoints HTTP. Dependências sugeridas: busboy (streaming multipart), csv-parse (parser streaming opcional), ou apenas readline nativo para CSV simples.

Estrutura mínima do projeto

Crie um projeto com package.json e instale as dependências. Exemplo mínimo de comandos:

mkdir csv-stream-workers && cd csv-stream-workers
npm init -y
npm i express busboy

Vamos usar a API nativa de worker_threads, então não precisa instalar nada extra para isso.

Considerações de segurança

Uploads abrirem uma superfície de ataque: limite tamanho máximo por upload, valide o tipo de conteúdo, autentique quem pode enviar e sanitize qualquer dado antes de inserir em banco. Evite expor a rota SSE sem autenticação quando necessário. Use HTTPS em produção para proteger SSE e uploads.

3) Tutorial guiado passo a passo

3.1 Exemplo errado: ler tudo em memória (anti-pattern)

Primeiro, o erro clássico. Abaixo um exemplo com multer memoryStorage que lê tudo e processa sincronicamente. Isso parece simples, mas é perigoso para arquivos grandes.

Contexto: este código aceita um upload e tenta parsear todo o CSV para memória usando split por newline.

// server-wrong.js
const express = require('express')
const multer = require('multer')
const upload = multer({ storage: multer.memoryStorage(), limits: { fileSize: 50 * 1024 * 1024 } })
const app = express()

app.post('/upload-wrong', upload.single('file'), (req, res) => {
  // Erro: carrega o arquivo inteiro na memória
  const buffer = req.file.buffer
  const content = buffer.toString('utf8')
  const rows = content.split('\n').map(r => r.split(','))

  // Simula trabalho pesado síncrono
  let processed = 0
  for (const row of rows) {
    // processamento pesado simulado
    processed += row.length
  }

  res.json({ rows: rows.length, processed })
})

app.listen(3000)

O que acontece: funciona para alguns MBs, mas se o arquivo tiver centenas de MBs ou GBs sua aplicação vai consumir muita memória e o event loop pode travar durante o parsing síncrono.

3.2 Por que é ruim (erros comuns e armadilhas)

Erros comuns ao seguir essa abordagem:

• memoryStorage aceita arquivos grandes e os coloca em Buffer na memória
• split('\n') cria um array com todas as linhas — grande overhead
• processamento síncrono bloqueia o event loop
• sem limites/validações, um atacante pode enviar vários arquivos e causar DoS

Dica: se você precisa de upload multi-GB, nunca use armazenamento em memória. Sempre use streaming para processar por partes.

3.3 Correção: receber o upload como stream

Agora o jeito certo: usar busboy (ou o parser de multipart do framework) para ler o arquivo como stream. Vou mostrar um endpoint que aceita o upload e encaminha o stream para um parser linha a linha usando readline. Isso evita armazenar tudo em memória.

Contexto: o código abaixo lê o arquivo enviado multipart/form-data e faz parse por linha do CSV com readline do Node.

// server-stream.js
const express = require('express')
const Busboy = require('busboy')
const readline = require('readline')
const { Worker } = require('worker_threads')
const { v4: uuidv4 } = require('uuid')

const app = express()
const jobs = new Map() // guarda estado para SSE

app.post('/upload', (req, res) => {
  const busboy = new Busboy({ headers: req.headers })
  const jobId = uuidv4()
  jobs.set(jobId, { processed: 0, total: 0 })

  busboy.on('file', (fieldname, file, filename) => {
    // file é um stream legível
    const rl = readline.createInterface({ input: file })
    const worker = new Worker('./worker.js')

    const batch = []
    const BATCH_SIZE = 1000

    rl.on('line', (line) => {
      jobs.get(jobId).total++
      batch.push(line)
      if (batch.length >= BATCH_SIZE) {
        worker.postMessage({ lines: batch.splice(0, batch.length) })
      }
    })

    rl.on('close', () => {
      if (batch.length) worker.postMessage({ lines: batch.splice(0, batch.length) })
      worker.postMessage({ done: true })
    })

    worker.on('message', (msg) => {
      if (msg.processed) {
        jobs.get(jobId).processed += msg.processed
      }
    })

    worker.on('exit', () => {
      // worker terminou
    })
  })

  busboy.on('finish', () => {
    res.json({ jobId })
  })

  req.pipe(busboy)
})

app.listen(3000)

O que acontece: o arquivo não é armazenado na memória. readline entrega linhas uma a uma. Enviamos batches para um Worker Thread que faz o processamento pesado em paralelo sem bloquear o event loop.

3.4 O Worker Thread: processamento sem bloquear

Contexto: o worker recebe batches de linhas, parseia e faz trabalho intensivo (ex: validação, transformação). O resultado pode ser armazenado em banco ou enviado de volta ao main thread como notificações de progresso.

// worker.js
const { parentPort } = require('worker_threads')

let processed = 0

parentPort.on('message', (msg) => {
  if (msg.lines) {
    for (const line of msg.lines) {
      // parse simples CSV (não robusto contra vírgulas em quotes)
      const cols = line.split(',')
      // processamento CPU-bound simulado
      processed += cols.length
    }
    parentPort.postMessage({ processed: msg.lines.length })
  }

  if (msg.done) {
    parentPort.postMessage({ done: true, processed })
    process.exit(0)
  }
})

O que acontece: o worker calcula e envia de volta quantas linhas processou. O main thread atualiza o estado do job (jobs Map) usado pelo SSE para notificar clientes.

3.5 Progresso em tempo real com SSE

Contexto: SSE (Server-Sent Events) é uma conexão unidirecional do servidor para o cliente que mantém um stream de eventos. É simples de implementar: o cliente abre uma EventSource e o servidor escreve linhas no formato "data: ...\n\n".

// server-sse.js (trecho do mesmo app)
app.get('/events/:jobId', (req, res) => {
  const jobId = req.params.jobId
  const state = jobs.get(jobId)
  if (!state) return res.status(404).end()

  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive'
  })

  const interval = setInterval(() => {
    const s = jobs.get(jobId)
    if (!s) return
    res.write(`data: ${JSON.stringify({ processed: s.processed, total: s.total })}\n\n`)
    if (s.total > 0 && s.processed >= s.total) {
      res.write(`data: ${JSON.stringify({ done: true })}\n\n`)
      clearInterval(interval)
      // opcional: jobs.delete(jobId)
      res.end()
    }
  }, 500)

  req.on('close', () => clearInterval(interval))
})

O que acontece: o cliente que abriu /events/:jobId recebe atualizações regulares sobre processed/total. Quando terminado, o servidor envia done e fecha a conexão.

4) Resultado final funcionando

Exemplo completo (server.js) — como juntar tudo

Aqui temos um resumo de como unir upload stream, Worker Threads e SSE numa única aplicação Express. Código reduzido, funcionando, pronto para testes locais.

// server-final.js (resumo funcional)
const express = require('express')
const Busboy = require('busboy')
const readline = require('readline')
const { Worker } = require('worker_threads')
const { v4: uuidv4 } = require('uuid')

const app = express()
const jobs = new Map()

app.post('/upload', (req, res) => {
  const busboy = new Busboy({ headers: req.headers })
  const jobId = uuidv4()
  jobs.set(jobId, { processed: 0, total: 0 })

  busboy.on('file', (fieldname, file) => {
    const rl = readline.createInterface({ input: file })
    const worker = new Worker('./worker.js')
    const batch = []
    const BATCH_SIZE = 1000

    rl.on('line', (line) => {
      const state = jobs.get(jobId)
      state.total++
      batch.push(line)
      if (batch.length >= BATCH_SIZE) worker.postMessage({ lines: batch.splice(0, BATCH_SIZE) })
    })

    rl.on('close', () => {
      if (batch.length) worker.postMessage({ lines: batch.splice(0, batch.length) })
      worker.postMessage({ done: true })
    })

    worker.on('message', (msg) => {
      if (msg.processed) jobs.get(jobId).processed += msg.processed
    })
  })

  busboy.on('finish', () => res.json({ jobId }))
  req.pipe(busboy)
})

app.get('/events/:jobId', (req, res) => {
  const jobId = req.params.jobId
  if (!jobs.has(jobId)) return res.status(404).end()

  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive'
  })

  const interval = setInterval(() => {
    const s = jobs.get(jobId)
    if (!s) return
    res.write(`data: ${JSON.stringify({ processed: s.processed, total: s.total })}\n\n`)
    if (s.total > 0 && s.processed >= s.total) {
      res.write(`data: ${JSON.stringify({ done: true })}\n\n`)
      clearInterval(interval)
      res.end()
    }
  }, 500)

  req.on('close', () => clearInterval(interval))
})

app.listen(3000)

O que acontece: esse arquivo combina as ideias mostradas antes. Em produção você deve tratar erros, logs, limites e autenticação.

Cliente JavaScript para SSE

Contexto: cliente abre EventSource para receber progresso e pode usar fetch para enviar o arquivo. Exemplo de cliente que mostra progresso no console.

// client.js (browser)
// enviar com fetch + FormData
async function upload(file) {
  const fd = new FormData()
  fd.append('file', file)
  const r = await fetch('/upload', { method: 'POST', body: fd })
  const { jobId } = await r.json()

  const es = new EventSource(`/events/${jobId}`)
  es.onmessage = (e) => {
    const data = JSON.parse(e.data)
    console.log('progress', data)
    if (data.done) es.close()
  }
}

O que acontece: o navegador mostra atualizações periódicas do servidor até a tarefa terminar. SSE reconecta automaticamente em desconexões curtas, mas atenção a autenticação e headers CORS.

Erros comuns ao integrar tudo

Lista de problemas que você pode encontrar ao montar o sistema:

• Não fechar worker corretamente — pode gerar vazamento de processos
• Enviar muitas mensagens pequenas entre main e worker — overhead de IPC
• Ler linhas com readline pode quebrar se o CSV tem quebras dentro de quotes — use parser robusto se necessário
• Não validar conteúdo do CSV antes de inserir em DB — vulnerabilidade de SQL injection/CSV injection

Dica: prefira batches maiores para reduzir mensagens IPC e medir latência entre envio e confirmação.

5) Desafio para o leitor praticar

Desafio 1 — adicionar validação e inserção em banco

Modifique o worker para validar cada linha segundo um esquema (por exemplo: id:number, email:string) e, para linhas válidas, acumulá-las para inserir em batch no PostgreSQL usando COPY ou inserções em lote. Meça o throughput.

Desafio 2 — CSV com campos entre aspas e vírgulas internas

Substitua o parse simples por um parser robusto (ex: csv-parse em modo stream) dentro do worker. Compare a performance e documente trade-offs entre robustez e velocidade.

Desafio 3 — autenticação SSE e reconexão

Proteja a rota /events/:jobId com JWT e implemente lógica de reconexão no cliente para recuperar o estado do job caso o cliente perca a conexão. Considere usar ETag/last-event-id se necessário.

Conclusao pratica

Resumo das escolhas técnicas

Para processar grandes CSV em Node.js sem travar a aplicação, combine: (1) leitura por streams para evitar carregar tudo em memória, (2) delegar CPU-bound para Worker Threads para não bloquear o event loop e (3) expor progresso com SSE para feedback em tempo real. Cada peça resolve um problema específico: memória, bloqueio e UX.

Checklist para produção

1. Limites de upload e autenticação nas rotas
2. Validação e escaping antes da persistência
3. Gestão de workers (pooling) e políticas de retry
4. Monitoramento de uso de memória e CPU

Implementar corretamente esses padrões reduz risco de incidentes e melhora a experiência do usuário. Comece pequeno: transforme o anti-pattern em streaming, depois adicione worker e SSE. Teste com arquivos reais e meça — sempre meça.

Se precisar, posso fornecer um repositório inicial completo com scripts de teste (gerador de CSV), configuração de pool de workers e integração com PostgreSQL. Peça o repositório e eu gero os arquivos prontos para rodar.