Gerar thumbnails e checksums de uploads usando Streams e Worker Threads no Node.js

Motivacao e trade-offs

Por que não carregar tudo na memória

Uploads de arquivos, especialmente imagens e vídeos, crescem em volume e tamanho. Ler um arquivo inteiro na memória para gerar uma thumbnail ou calcular um checksum é uma rota simples, mas perigosa: aumenta o consumo de memória, cria picos GC e torna a aplicação fragilizada frente a ataques de negação de serviço. Para APIs reais é necessário processar fluxos de dados de forma linear e controlar o uso de CPU e memória.

Quando usar worker threads

Operações como compressão, redimensionamento de imagem (libvips/sharp) e hashing de grandes buffers são CPU-bound. Node.js é single-threaded no loop de eventos; ao executar tarefas pesadas no processo principal você atrasa handling de conexões e timers. Worker threads permitem mover o trabalho pesado para threads separadas mantendo o loop de evento responsivo. O trade-off é a complexidade de comunicação e a cópia/transferência de memória entre threads.

As decisões práticas envolvem: (1) streaming para reduzir RAM, (2) duplicação controlada do stream quando precisar de múltiplos consumidores (hash + disco + processamento), (3) offload para workers quando CPU for o gargalo. Em alguns cenários faz sentido combinar: stream para armazenamento imediato + checksum no main thread + thumbnail no worker.

Dica: prefira pipelines e PassThroughs em vez de buffers intermediários. Use arquivos temporários quando o consumo de memória para reconstruir o stream precisar ser evitado.

Fundamentos tecnicos

Streams: duplicar e pipeline

Streams do Node servem para processar dados chunk a chunk. Para alimentar múltiplos consumidores do mesmo upload é comum criar PassThroughs e pipear a origem para várias saídas. Exemplo de padrão: criar duas PassThroughs e direcionar uma para hashing e outra para processamento da thumbnail. Use stream.pipeline (ou a versão promise) para lidar corretamente com erros e fechamento das streams.

Worker Threads: postar mensagens vs transferir ArrayBuffers

Worker threads se comunicam via postMessage. Para evitar cópia de grandes buffers, transfira um ArrayBuffer via transfer list. Quando você tiver um Buffer do Node, convém extrair a porção exata do ArrayBuffer com buffer.buffer.slice(byteOffset, byteOffset + length) para transferir sem surpresas.

Segurança e validação

Antes de processar: valide extensão, tipo MIME (mimetypes são fáceis de falsificar), limite o tamanho do upload, verifique dimensões esperadas e aplique um scanner de malware em uploads críticos. Não confie apenas no nome do arquivo enviado pelo cliente.

Exemplo 1: caso simples

O que vamos fazer

Implementar um endpoint Express que aceita upload via Busboy, grava o arquivo em disco, gera uma thumbnail com sharp e calcula um SHA-256 sem carregar tudo na memória. A estratégia: duplicar o stream com PassThrough e usar pipelines separados.

Código: servidor Express (funcional)

Contexto: este endpoint usa busboy, sharp, stream e crypto. Ele grava o upload em uploads/, calcula o checksum e gera uma thumbnail 200px wide.

const express = require('express');
const fs = require('fs');
const path = require('path');
const Busboy = require('busboy');
const { pipeline, PassThrough, Writable } = require('stream');
const { promisify } = require('util');
const crypto = require('crypto');
const sharp = require('sharp');

const pipe = promisify(pipeline);
const app = express();

app.post('/upload', (req, res) => {
  const busboy = new Busboy({ headers: req.headers });
  let responded = false;

  busboy.on('file', (fieldname, file, info) => {
    const filename = path.basename(info.filename);
    const uploadPath = path.join(__dirname, 'uploads', filename);

    // duplicadores de stream
    const passForHash = new PassThrough();
    const passForSharp = new PassThrough();

    // encaixar duplicação
    file.pipe(passForHash);
    file.pipe(passForSharp);

    // sink para salvar arquivo
    const writeStream = fs.createWriteStream(uploadPath);

    // hash writable
    const hash = crypto.createHash('sha256');
    const hashSink = new Writable({
      write(chunk, enc, cb) { hash.update(chunk); cb(); }
    });

    // pipeline para salvar e criar thumbnail
    Promise.all([
      pipe(passForHash, hashSink),
      pipe(passForSharp, sharp().resize(200).jpeg(), fs.createWriteStream(uploadPath + '.thumb.jpg')),
      pipe(file, writeStream) // garante que file também vá para disco (opcional)
    ]).then(() => {
      const digest = hash.digest('hex');
      if (!responded) { responded = true; res.json({ ok: true, filename, sha256: digest }); }
    }).catch(err => {
      if (!responded) { responded = true; res.status(500).json({ error: err.message }); }
    });
  });

  busboy.on('error', err => res.status(500).json({ error: err.message }));
  req.pipe(busboy);
});

app.listen(3000);

O que acontece: o arquivo enviado é duplicado para hashing e para o pipeline do sharp. Usamos pipelines separados para automatizar o tratamento de erros. Evitamos armazenar o arquivo completo na memória.

Exemplo 2: caso avancado do mundo real

O desafio real

Em produção você pode ter uploads grandes e múltiplas transformações (várias resoluções de thumbnails, WebP conversion, validação e assinatura de conteúdo). Aqui é preferível gravar o upload em disco (ou S3 multipart) e delegar o processamento pesado para uma pool de workers, controlando concorrência.

Estratégia adotada

Nesta abordagem: (1) stream para um arquivo temporário enquanto calculamos um checksum rápido; (2) ao finalizar, enviamos o caminho para um worker thread que abre o arquivo, executa sharp para gerar todas as thumbnails e retorna buffers transferidos de volta (ou escreve em um storage externo).

Código: main thread (upload + delegação ao worker)

Contexto: este script grava em disco e posta a tarefa para um worker localizado em worker.js. O main thread mantém a responsividade e só espera o trabalho assíncrono terminar para atualizar o banco.

// main.js
const express = require('express');
const fs = require('fs');
const path = require('path');
const Busboy = require('busboy');
const crypto = require('crypto');
const { Worker } = require('worker_threads');
const { pipeline, PassThrough, Writable } = require('stream');
const { promisify } = require('util');
const pipe = promisify(pipeline);

const app = express();

app.post('/upload', (req, res) => {
  const busboy = new Busboy({ headers: req.headers });

  busboy.on('file', (fieldname, file, info) => {
    const filename = `${Date.now()}-${path.basename(info.filename)}`;
    const tmpPath = path.join(__dirname, 'tmp', filename);
    const writeStream = fs.createWriteStream(tmpPath);

    const hash = crypto.createHash('sha256');
    const hashSink = new Writable({ write(chunk, enc, cb) { hash.update(chunk); cb(); } });

    // duplicar stream para salvar + hash
    const passForHash = new PassThrough();
    file.pipe(passForHash);
    file.pipe(writeStream);

    pipe(passForHash, hashSink).then(() => {
      const digest = hash.digest('hex');
      // Delegar para worker
      const worker = new Worker(path.join(__dirname, 'worker.js'));
      worker.once('message', msg => {
        // msg => { thumbs: [{name, buffer, length}], info }
        // reconstrói buffers a partir de ArrayBuffers transferidos
        for (const t of msg.thumbs) {
          const ab = t.buffer; // já é ArrayBuffer
          const buf = Buffer.from(ab);
          fs.writeFileSync(path.join(__dirname, 'uploads', t.name), buf);
        }
        fs.unlinkSync(tmpPath);
        res.json({ ok: true, sha256: digest, thumbs: msg.thumbs.map(t => t.name) });
        worker.terminate();
      });

      worker.postMessage({ filePath: tmpPath, tasks: [{ width: 200, suffix: '200' }, { width: 800, suffix: '800' }] });
    }).catch(err => {
      res.status(500).json({ error: err.message });
    });
  });

  req.pipe(busboy);
});

app.listen(3000);

O que acontece: o upload é salvo em tmp; em paralelo calculamos o SHA-256. Quando o hash finaliza, criamos um worker e pedimos para gerar thumbs. Usamos escrita em disco no worker para reduzir a transferência de dados entre threads, mas também mostramos como transferir buffers se necessário.

Código: worker (worker.js)

Contexto: o worker abre o arquivo no disco, gera thumbnails usando sharp e envia os buffers de volta ao main thread como ArrayBuffers transferidos para evitar cópia desnecessária.

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

parentPort.on('message', async (msg) => {
  const { filePath, tasks } = msg;
  const results = [];
  try {
    for (const t of tasks) {
      const buf = await sharp(filePath).resize({ width: t.width }).toFormat('jpeg').toBuffer();
      // extrai ArrayBuffer exato para transferir
      const ab = buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.length);
      results.push({ name: `${path.basename(filePath)}.${t.suffix}.jpg`, buffer: ab, length: buf.length });
    }
    parentPort.postMessage({ thumbs: results });
  } catch (err) {
    parentPort.postMessage({ error: err.message });
  }
});

O que acontece: o worker gera os buffers e os envia de volta. No main thread reconstruímos os Buffers a partir do ArrayBuffer e gravamos os arquivos finais. Isso evita que o main thread faça processamento de imagem e mantém o loop de eventos livre.

Debugging e troubleshooting

Erros comuns e armadilhas

Listagem dos problemas que aparecem com frequência:

• Perda de dados ao pipear múltiplos consumidores sem PassThrough; resultado: apenas um consumidor recebe dados.
• Fazer Buffer.from(ab) com um ArrayBuffer maior que a fatia correta — gere resultados corrompidos se não slicear corretamente o ArrayBuffer.
• Esquecer de lidar com erros em pipelines: um erro em uma das streams pode deixar a requisição pendente. Use stream.pipeline ou o evento 'error'.
• Registrar thumbnails grandes sem controlar concorrência — em picos isso pode criar fila de tasks e exaurir CPU.

Logs e introspecção

Como debugar: adicione logs estruturados nos pontos de entrada/saída das etapas: início do upload, término de gravação em tmp, envio ao worker, término do worker. Em workers, capture exceções não tratadas com process.on('uncaughtException') para enviar informação de volta ao main thread antes de terminar. Use métricas (latência por upload, uso de CPU por worker) para ajustar pool de workers.

Dicas práticas

Pool de workers: crie uma pool limitada em vez de instanciar um worker por upload para evitar escalonamento excessivo. Uma pool de tamanho igual ao número de CPUs ou menos costuma funcionar bem. Gerencie fila de tarefas com backpressure.

Dica: prefira escrever thumbails direto para storage externo (S3) a partir do worker quando possível — isso elimina cópias adicionais e passo de disco no main thread.

Conclusão prática

Resumo das recomendações

Combine streams e worker threads: use streams para reduzir uso de memória e duplicar dados via PassThrough apenas quando necessário; use worker threads para trabalhos CPU-bound como processamento de imagens; prefira transferência de ArrayBuffers ao enviar grandes dados entre threads para minimizar cópias.

Checklist antes de produção

1. Limitar tamanho máximo de upload e tempo máximo por upload.
2. Validar tipo de arquivo e aplicar verificação adicional (magic bytes).
3. Ter pool de workers e limites de concorrência para thumbnails.
4. Registrar métricas de latência e uso de recursos.

Implementações que respeitam esses pontos mantêm a API responsiva e segura, mesmo sob carga. A arquitetura combinando streams com worker threads é prática, escalável e relativamente simples de operar quando estruturada por pipelines e com boa observabilidade.

Este artigo apresentou padrões aplicáveis ao dia a dia: duplicação de streams com PassThrough, pipelines com tratamento de erro, gravação em disco como fallback e offload de CPU usando worker threads com transferência controlada de ArrayBuffers. Esses padrões reduzem riscos de OOM, mantêm o loop responsivo e permitem processar uploads de forma robusta.