Practice — Upload Endpoint with Queue — Node.js & Express Backend

Practice: Upload Endpoint with Queue

This exercise combines file uploads, cloud storage, and job queues into a realistic production workflow. You will build a product image upload endpoint that: accepts an image via Multer (memoryStorage), adds an image-processing job to a BullMQ queue, returns 202 Accepted immediately, and processes the image asynchronously in a worker (resize via Sharp, upload to Cloudinary, update the database).

Analogy🏏Cricket

🏏 Think of it like cricket: this exercise is your full match-day rehearsal, stitching three drilled skills into one seamless play. Multer's memoryStorage catches the incoming file the way a wicketkeeper cleanly gloves the ball the instant it arrives. Rather than making the batsman wait while you polish and mount that ball, you immediately add an image-processing job to the BullMQ queue and return 202 Accepted — like the keeper flicking the ball to a fielder and getting straight back into position, telling the batsman 'received, play on'. Then, off to the side, the worker does the slow craft: Sharp resizes the image, Cloudinary stores it, the database is updated — just as the boundary rider takes his time to field, relay, and let the scorer record it, all without holding up the next delivery. The payoff: rehearsing upload, queue, and async worker together builds the exact production reflex where users get an instant response while heavy processing happens off the field.

What You'll Build

POST /api/products/:id/image — accepts a multipart/form-data request, validates the file, adds a job to the imageQueue, and returns 202 with a jobId. A separate ImageWorker resizes the image to 800x800 (Sharp), uploads to Cloudinary, and updates the product's imageUrl in MongoDB. GET /api/jobs/:id returns the job's current status.

Analogy🏏Cricket

🏏 Think of it like cricket: You are the data operations manager at the BCCI analytics division, responsible for processing the complete ball-by-ball feed from an IPL season — 74 matches, 888 overs, over 5,000 deliveries. The raw feed arrives as a continuous data stream from the scoring tablets at each ground. Your job is to build the pipeline that ingests that stream, filters out extras and wide deliveries for certain statistics, aggregates the useful data into player-level summaries, and publishes the final report to the BCCI's official statistics portal before the next morning's press conference. Just as the BCCI would never ask analysts to hold the entire season's data in memory before starting analysis (they process each match's data as it arrives from the ground), your pipeline processes each CSV chunk as it streams from disk — maintaining a constant memory footprint regardless of how many seasons of data you process.

Prerequisites

Completed lessons 25–29 (Multer, Cloudinary, Nodemailer, BullMQ, Cron)
Node.js 18+, Redis running locally (redis-server or Docker)
A Cloudinary account (free tier) and MongoDB instance
npm packages from lessons 25–28

Step 1: Project Setup

bash

mkdir upload-queue-app && cd upload-queue-app
npm init -y
npm install express multer sharp cloudinary streamifier bullmq ioredis mongoose dotenv

javascript

// .env
PORT=3000
MONGODB_URI=mongodb://localhost:27017/upload_queue
REDIS_HOST=localhost
CLOUDINARY_CLOUD_NAME=your_cloud_name
CLOUDINARY_API_KEY=your_api_key
CLOUDINARY_API_SECRET=your_api_secret

Step 2: Queue and Cloudinary Config

javascript

// queues/imageQueue.js
const { Queue } = require('bullmq');
const connection = { host: process.env.REDIS_HOST || 'localhost', port: 6379 };
module.exports = new Queue('image-processing', { connection });

// config/cloudinary.js
const cloudinary = require('cloudinary').v2;
cloudinary.config({
  cloud_name: process.env.CLOUDINARY_CLOUD_NAME,
  api_key:    process.env.CLOUDINARY_API_KEY,
  api_secret: process.env.CLOUDINARY_API_SECRET
});
module.exports = cloudinary;

// models/Product.js
const mongoose = require('mongoose');
const productSchema = new mongoose.Schema({
  name:        String,
  price:       Number,
  imageUrl:    { type: String, default: null },
  cloudinaryId:{ type: String, default: null },
  imageStatus: { type: String, enum: ['none','processing','ready'], default: 'none' }
}, { timestamps: true });
module.exports = mongoose.model('Product', productSchema);

Step 3: Upload Route

javascript

// routes/products.js
const router      = require('express').Router();
const multer      = require('multer');
const Product     = require('../models/Product');
const imageQueue  = require('../queues/imageQueue');

const upload = multer({
  storage: multer.memoryStorage(),
  limits:  { fileSize: 5 * 1024 * 1024 },
  fileFilter: (req, file, cb) => {
    const allowed = ['image/jpeg', 'image/png', 'image/webp'];
    allowed.includes(file.mimetype) ? cb(null, true) : cb(new Error('Invalid image type'));
  }
});

// POST /api/products/:id/image
router.post('/:id/image', upload.single('image'), async (req, res) => {
  const product = await Product.findById(req.params.id);
  if (!product) return res.status(404).json({ error: 'Product not found' });
  if (!req.file)  return res.status(400).json({ error: 'No file uploaded' });

  // Mark product as processing
  await Product.updateOne({ _id: product.id }, { imageStatus: 'processing' });

  // Add job to queue with image buffer (as base64 for Redis serialisation)
  const job = await imageQueue.add('process-image', {
    productId:    product.id.toString(),
    imageBuffer:  req.file.buffer.toString('base64'),
    mimetype:     req.file.mimetype
  }, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 3000 }
  });

  // Return 202 Accepted — processing is happening in the background
  res.status(202).json({
    message:   'Image upload accepted, processing in background',
    jobId:     job.id,
    statusUrl: '/api/jobs/' + job.id
  });
});

module.exports = router;

Step 4: Image Worker

javascript

// workers/imageWorker.js
require('dotenv').config();
const { Worker } = require('bullmq');
const sharp      = require('sharp');
const streamifier = require('streamifier');
const cloudinary = require('../config/cloudinary');
const Product    = require('../models/Product');
const mongoose   = require('mongoose');

mongoose.connect(process.env.MONGODB_URI);

const connection = { host: process.env.REDIS_HOST || 'localhost', port: 6379 };

function uploadToCloudinary(buffer, options) {
  return new Promise((resolve, reject) => {
    const stream = cloudinary.uploader.upload_stream(options, (err, result) => {
      err ? reject(err) : resolve(result);
    });
    streamifier.createReadStream(buffer).pipe(stream);
  });
}

const imageWorker = new Worker('image-processing', async (job) => {
  const { productId, imageBuffer, mimetype } = job.data;

  console.log('[ImageWorker] Processing job', job.id, 'for product', productId);

  // Decode base64 buffer
  const inputBuffer = Buffer.from(imageBuffer, 'base64');

  // Resize and compress with Sharp
  const processedBuffer = await sharp(inputBuffer)
    .resize(800, 800, { fit: 'inside', withoutEnlargement: true })
    .jpeg({ quality: 85 })
    .toBuffer();

  // Upload to Cloudinary
  const result = await uploadToCloudinary(processedBuffer, {
    folder:    'products',
    public_id: 'product_' + productId,
    overwrite: true
  });

  // Update product in DB
  await Product.updateOne(
    { _id: productId },
    {
      imageUrl:     result.secure_url,
      cloudinaryId: result.public_id,
      imageStatus:  'ready'
    }
  );

  console.log('[ImageWorker] Done. URL:', result.secure_url);
  return { imageUrl: result.secure_url };
}, { connection, concurrency: 2 });

imageWorker.on('failed', async (job, err) => {
  console.error('[ImageWorker] Failed:', job.id, err.message);
  await Product.updateOne({ _id: job.data.productId }, { imageStatus: 'none' });
});

Step 5: Job Status Endpoint and App Wiring

javascript

// routes/jobs.js
const router     = require('express').Router();
const { Queue }  = require('bullmq');
const connection = { host: process.env.REDIS_HOST || 'localhost', port: 6379 };
const imageQueue = new Queue('image-processing', { connection });

router.get('/:id', async (req, res) => {
  const job = await imageQueue.getJob(req.params.id);
  if (!job) return res.status(404).json({ error: 'Job not found' });

  const state = await job.getState(); // waiting | active | completed | failed | delayed

  res.json({
    id:       job.id,
    state,
    progress: job.progress,
    result:   state === 'completed' ? job.returnvalue : null,
    reason:   state === 'failed'    ? job.failedReason : null
  });
});

module.exports = router;

// app.js
require('dotenv').config();
const express  = require('express');
const mongoose = require('mongoose');
const app      = express();

app.use(express.json());
app.use('/api/products', require('./routes/products'));
app.use('/api/jobs',     require('./routes/jobs'));

app.use((err, req, res, next) => {
  if (err.message?.includes('Invalid image type'))
    return res.status(415).json({ error: err.message });
  console.error(err);
  res.status(500).json({ error: 'Server error' });
});

mongoose.connect(process.env.MONGODB_URI).then(() => {
  app.listen(process.env.PORT, () => console.log('Upload API on port', process.env.PORT));
});

bash

# Terminal 1: start API server
node app.js

# Terminal 2: start image worker
node workers/imageWorker.js

# Create a test product
PRODUCT_ID=$(curl -s -X POST http://localhost:3000/api/products \
  -H "Content-Type: application/json" \
  -d '{"name":"Cricket Bat","price":120}' | jq -r ._id)

# Upload image
JOB_ID=$(curl -s -X POST \
  http://localhost:3000/api/products/$PRODUCT_ID/image \
  -F "image=@/path/to/photo.jpg" | jq -r .jobId)

# Poll job status
curl http://localhost:3000/api/jobs/$JOB_ID

Image buffers encoded as base64 in job payloads inflate the Redis memory usage by ~33%. For large images (>2 MB), save the buffer to a temporary S3 object or a local temp file, pass only the key in the job payload, and clean up after the worker finishes.

The imageStatus field on the Product model lets the frontend poll the product endpoint or subscribe to a WebSocket event instead of tracking job IDs. Update imageStatus to 'ready' in the worker's completion handler, and to 'none' on failure.

Use multer memoryStorage to keep the file as a Buffer before passing to the queue
Encode Buffer as base64 for Redis-safe job payload serialisation
Return 202 Accepted immediately; the client polls /api/jobs/:id for status
Sharp resizes and compresses in one pass before Cloudinary upload
Workers run in a separate process; restart them independently from the API server
Mark imageStatus on the resource model so the frontend can react without tracking job IDs