Architecting Resilient Ingestion

The Problem: The WebSocket Trap

In modern application design, the default reflex for bidirectional communication is often to just open a WebSocket. While brilliant for low-latency text streams, using WebSockets as a catch-all pipeline for high-payload data creates massive architectural bottlenecks known as Head-of-Line (HoL) Blocking.

To eliminate these silent failures, I engineered a resilient ingestion workflow that decoupled heavy payload data from the real-time stream. We separated network traffic into two distinct lanes based on payload weight.

1. The Hybrid Frontend Client

The React client dictates the routing. Sub-millisecond text messages and state updates flow through the socket.io instance. However, when a user drops a file into the chat, the application bypasses the socket entirely. It requests a presigned URL from the backend and pushes the heavy binary data directly to MinIO using a standard XMLHttpRequest, allowing for real-time progress tracking without clogging the TCP pipeline.

const uploadFileWithProgress = (file: File, url: string, messageId: string): Promise<void> => {
  return new Promise((resolve, reject) => {
    const xhr = new XMLHttpRequest();
    xhr.open('PUT', url, true);
    
    xhr.upload.onprogress = (event) => {
      if (event.lengthComputable) {
        const percentComplete = Math.round((event.loaded / event.total) * 100);
        setMessages((prev) => prev.map(msg => 
          msg.id === messageId ? { ...msg, progress: percentComplete } : msg
        ));
      }
    };

    xhr.onload = () => {
      if (xhr.status >= 200 && xhr.status < 300) {
        setMessages((prev) => prev.map(msg => 
          msg.id === messageId ? { ...msg, status: 'ready', progress: 100 } : msg
        ));
        resolve();
      } else {
        reject(new Error(`Upload failed with status ${xhr.status}`));
      }
    };

    xhr.onerror = () => reject(new Error('Network Error during upload'));
    xhr.send(file);
  });
};

2. The Decoupled API & Message Queue

Routing heavy files through REST solves network drops, but synchronous processing would block the Node.js event loop. The Express API is strictly responsible for orchestration. It issues MinIO presigned URLs and acts as a producer for RabbitMQ when heavy background tasks (like zipping multiple files) are requested. It instantly returns a 202 Accepted.

app.post('/api/upload-url', async (req: Request, res: Response) => {
  try {
    const { fileName } = req.body;
    if (!fileName) return res.status(400).json({ error: 'fileName is required' });

    const url = await generatePresignedUploadUrl(fileName);
    return res.json({ url });
  } catch (error) {
    return res.status(500).json({ error: 'Internal server error' });
  }
});

app.post('/api/request-zip', async (req: Request, res: Response) => {
  try {
    const { room, files } = req.body;
    if (!room || !Array.isArray(files) || files.length === 0) {
      return res.status(400).json({ error: 'Invalid payload' });
    }

    const jobId = `job_${Date.now()}`;
    await publishZipJob(room, files, jobId);

    return res.status(202).json({ message: 'Zip job queued', jobId });
  } catch (error) {
    return res.status(500).json({ error: 'Internal server error' });
  }
});

3. Stream-Optimized Background Workers

The dedicated worker service listens to RabbitMQ with a strict channel.prefetch(1). This prevents memory exhaustion during traffic spikes by forcing the worker to process only one heavy job at a time.

Crucially, the file zipping is completely stream-based. Instead of downloading all files into memory or writing them to a local disk, I used a Node.js PassThrough stream as a bridge. The archiver library compresses the incoming file streams on the fly and immediately pipes the output buffer straight back into MinIO.

const processZipJob = async (files: string[], jobId: string): Promise<string> => {
  const zipFileName = `zips/${jobId}.zip`;
  
  // PassThrough acts as a bridge between the archiver output and MinIO input
  const streamBridge = new PassThrough();
  const archive = archiver('zip', { zlib: { level: 9 } });

  archive.on('error', (err) => { throw err; });
  archive.pipe(streamBridge);

  for (const file of files) {
    const fileStream = await getFileStream(file);
    archive.append(fileStream, { name: file });
  }

  archive.finalize();
  await uploadStream(zipFileName, streamBridge);

  return zipFileName;
};

// ... RabbitMQ Consumer Setup ...
channel.consume(config.rabbitmq.jobQueue, async (msg) => {
  if (!msg) return;
  const { room, files, jobId } = JSON.parse(msg.content.toString());

  try {
    const zipFileName = await processZipJob(files, jobId);
    
    // Notify the backend API that the zip is ready
    const resultPayload = JSON.stringify({ room, jobId, zipFileName });
    channel.sendToQueue(config.rabbitmq.resultQueue, Buffer.from(resultPayload), {
      persistent: true,
    });

    // Acknowledge the message only after complete success
    channel.ack(msg);
  } catch (error) {
    // Reject message without requeuing to prevent infinite crash loops
    channel.nack(msg, false, false);
  }
});

Conclusion