Node Express BullMQ Redis statement generator
A statement desk turns up to 200 customer balance rows into individual PDFs and one downloadable ZIP while showing live progress. Tokay runs the upload page and BullMQ worker separately over one managed Redis resource.
Last updated
This is the pattern to use when a request should start substantial work without making the visitor wait on one web process. Redis carries the queue, progress, CSV, and temporary PDFs between both Services.
- Runtime
- Node.js
- Framework
- Express and BullMQ
- Service types
- Web Service, Background Worker
- Resources
- Redis
- Required secrets
- None
What it does
The page offers a 40 row sample CSV and accepts your own file with customer, email, period, and balance columns.
Press Generate statements and the current batch counter advances as the Worker finishes each PDF. Completed names become individual download links, and Download all PDFs appears when the full batch is ready.
The desk keeps the newest eight completed batches. A ninth removes the oldest temporary source and output files from Redis.
Deploy it
Bring this folder to Tokay any way you like. Paste or upload it in the dashboard, push it with git, or hand it to your AI agent along with our agent instructions. Tokay figures out the rest.
Working in Claude, ChatGPT, VS Code, or another MCP client? Connect the Tokay MCP server and ask your agent to deploy this example. It signs in through your browser, so there is no token to paste.
New to Tokay? The getting started guide walks you through your first deploy.
Try it
- Download the sample CSV, upload it, and watch the counter reach 40.
- Open one customer statement and confirm that the balance appears in a valid PDF.
- Download the ZIP and check that it contains all 40 named statement files.
Watch recovery
Start a larger batch, then stop the Background Worker while it is moving. Tokay restarts the process, and BullMQ returns the interrupted job to the queue. Watch the page continue from where the batch left off.
This learning app keeps the newest eight batches. Creating a ninth removes the oldest CSV and PDFs from Redis, so do not use the example as a permanent statement archive.
How it works
You keep the page responsive because the Web Service validates the CSV and adds one BullMQ job per customer instead of rendering PDFs itself. The Worker processes two at a time and saves each result to Redis with an idempotent completion marker.
The page polls batch metadata and serves individual files or a streamed ZIP. When statements become business records, move them to permanent object storage rather than stretching this intentionally short Redis history.
Secrets
Neither Service needs a secret from you. Tokay supplies the same private REDIS_URL to both of them.
Grow from here
Try the WebSocket chat when a long running process should coordinate connected browsers instead of queued work.
Source code
web/index.html Raw
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
<title>Northline statement desk</title>
<style>
:root{color-scheme:light;font-family:Inter,ui-sans-serif,system-ui,sans-serif;background:#efece5;color:#252925}*{box-sizing:border-box}body{margin:0}header{width:min(1080px,calc(100% - 36px));margin:auto;padding:22px 0;display:flex;justify-content:space-between;border-bottom:1px solid #c8c1b5;font-weight:800}main{width:min(1080px,calc(100% - 36px));margin:60px auto 90px}.hero{display:grid;grid-template-columns:1.25fr .75fr;gap:55px;align-items:end}.kicker{color:#a34c35;font-size:.75rem;font-weight:800;letter-spacing:.12em;text-transform:uppercase}h1{margin:10px 0 0;font:500 clamp(4rem,10vw,8rem)/.84 Georgia,serif;letter-spacing:-.06em}.hero>p{color:#65655e;font:1.1rem/1.7 Georgia,serif}.desk{display:grid;grid-template-columns:.8fr 1.2fr;gap:24px;margin-top:55px}.panel{padding:25px;border-radius:18px;background:#fbf8f1;box-shadow:0 18px 50px #453f3615}h2{font:500 2rem Georgia,serif}label{display:block;padding:30px 18px;border:2px dashed #b9b0a2;border-radius:12px;text-align:center}input{max-width:100%}button,.download{display:inline-block;margin-top:15px;padding:13px 18px;border:0;border-radius:9px;background:#31594d;color:white;font:inherit;font-weight:800;text-decoration:none;cursor:pointer}.sample{display:block;margin-top:15px;color:#31594d}.progress{height:12px;margin:22px 0;border-radius:10px;background:#ddd5c8;overflow:hidden}.progress span{display:block;width:0;height:100%;background:#c56945}.count{font:500 3rem Georgia,serif}.items{display:grid;gap:8px;margin-top:20px}.items a{display:flex;justify-content:space-between;padding:11px 0;border-top:1px solid #ded7cc;color:inherit;text-decoration:none}.error{color:#9a342a}@media(max-width:720px){.hero,.desk{grid-template-columns:1fr}}
</style>
</head>
<body>
<header><span>Northline Billing</span><span>Statement desk</span></header>
<main>
<section class="hero"><div><p class="kicker">A queue you can watch</p><h1>Forty statements. One upload.</h1></div><p>Upload customer balances and watch the worker turn each row into a finished PDF. Redis is the handoff and the temporary file cabinet.</p></section>
<section class="desk">
<form id="upload" class="panel"><h2>Start a batch</h2><label>Choose a CSV<input name="statements" type="file" accept=".csv,text/csv" required></label><button>Generate statements</button><a class="sample" href="/sample.csv">Download the 40 row sample</a><p id="error" class="error"></p></form>
<div class="panel"><h2>Current batch</h2><div id="count" class="count">Nothing queued</div><div class="progress"><span id="bar"></span></div><p id="status">Upload the sample to see real work move through BullMQ.</p><a id="zip" class="download" hidden>Download all PDFs</a><div id="items" class="items"></div></div>
</section>
</main>
<script>
const form=document.querySelector('#upload'),error=document.querySelector('#error'),count=document.querySelector('#count'),bar=document.querySelector('#bar'),status=document.querySelector('#status'),items=document.querySelector('#items'),zip=document.querySelector('#zip');
let timer;
function renderItems(batchId,batchItems){items.replaceChildren(...batchItems.slice(-8).reverse().map(item=>{const link=document.createElement('a'),name=document.createElement('span'),kind=document.createElement('strong');link.href=`/batches/${batchId}/statements/${item.index}.pdf`;name.textContent=item.customer;kind.textContent='PDF';link.append(name,kind);return link}))}
async function refresh(batchId){const response=await fetch(`/api/batches/${batchId}`);const batch=await response.json();if(!response.ok){error.textContent=batch.error;clearInterval(timer);return}const finished=batch.completed+batch.failed;count.textContent=`${batch.completed} / ${batch.total}`;bar.style.width=`${finished/batch.total*100}%`;status.textContent=batch.failed?`${batch.failed} statements could not be rendered.`:batch.completed===batch.total?'Every statement is ready.':`${batch.total-batch.completed} statements still in the queue.`;renderItems(batchId,batch.items);if(finished===batch.total){clearInterval(timer);if(batch.completed===batch.total){zip.href=`/batches/${batchId}/all.zip`;zip.hidden=false}}}
form.addEventListener('submit',async event=>{event.preventDefault();error.textContent='';zip.hidden=true;items.innerHTML='';const response=await fetch('/batches',{method:'POST',body:new FormData(form)});const result=await response.json();if(!response.ok){error.textContent=result.error;return}await refresh(result.batchId);timer=setInterval(()=>refresh(result.batchId),700)});
</script>
</body>
</html>
web/package.json Raw
{
"name": "statement-generator-web",
"version": "1.0.0",
"private": true,
"engines": { "node": ">=24" },
"scripts": { "start": "node server.js" },
"dependencies": {
"archiver": "8.0.0",
"bullmq": "5.80.9",
"csv-parse": "7.0.1",
"express": "5.2.1",
"ioredis": "5.11.1",
"multer": "2.2.0"
}
}
web/server.js Raw
const crypto = require("node:crypto");
const { ZipArchive } = require("archiver");
const { Queue } = require("bullmq");
const { parse } = require("csv-parse/sync");
const express = require("express");
const IORedis = require("ioredis");
const multer = require("multer");
const redis = new IORedis(process.env.REDIS_URL, { maxRetriesPerRequest: null });
const queue = new Queue("statements", { connection: redis });
const upload = multer({ storage: multer.memoryStorage(), limits: { fileSize: 256 * 1024 } });
const app = express();
const maxBatches = 8;
async function pruneBatches() {
let count = await redis.zcard("statement:batches");
if (count <= maxBatches) return;
const oldestFirst = await redis.zrange("statement:batches", 0, -1);
for (const batchId of oldestFirst) {
if (count <= maxBatches) break;
const metadata = await redis.hgetall(`statement:batch:${batchId}`);
if (Number(metadata.completed) + Number(metadata.failed) !== Number(metadata.total)) continue;
const keys = await redis.smembers(`statement:keys:${batchId}`);
if (keys.length) await redis.del(...keys);
await redis.del(`statement:keys:${batchId}`, `statement:batch:${batchId}`);
await redis.zrem("statement:batches", batchId);
count -= 1;
}
}
function sampleCsv() {
const names = ["Mira Chen", "Jon Bell", "Ada Evans", "Ravi Shah", "Nia Carter", "Luis Park", "Tess Morgan", "Omar Reed"];
const rows = ["customer,email,period,balance_cents"];
for (let index = 0; index < 40; index += 1) {
const name = names[index % names.length];
rows.push(`${name} ${Math.floor(index / names.length) + 1},customer${index + 1}@example.test,June 2026,${7400 + index * 315}`);
}
return `${rows.join("\n")}\n`;
}
app.get("/health", (_request, response) => response.json({ ok: true }));
app.get("/sample.csv", (_request, response) => {
response.type("text/csv").attachment("sample-statements.csv").send(sampleCsv());
});
app.post("/batches", upload.single("statements"), async (request, response, next) => {
try {
if (!request.file) return response.status(400).json({ error: "Choose a CSV file." });
const records = parse(request.file.buffer, { columns: true, skip_empty_lines: true, trim: true });
if (!records.length || records.length > 200) return response.status(400).json({ error: "Include between 1 and 200 statement rows." });
const rows = records.map((row, index) => {
const balanceCents = Number(row.balance_cents);
if (!row.customer || !row.email || !row.period || !Number.isInteger(balanceCents) || balanceCents < 0) {
throw new Error(`Row ${index + 2} needs customer, email, period, and a whole nonnegative balance_cents value.`);
}
return { index, customer: row.customer, email: row.email, period: row.period, balanceCents };
});
const batchId = crypto.randomUUID();
const metadataKey = `statement:batch:${batchId}`;
const csvKey = `statement:csv:${batchId}`;
await redis.hset(metadataKey, {
id: batchId,
total: rows.length,
completed: 0,
failed: 0,
createdAt: new Date().toISOString(),
filename: request.file.originalname,
});
await redis.set(csvKey, request.file.buffer);
await redis.sadd(`statement:keys:${batchId}`, metadataKey, csvKey);
await redis.zadd("statement:batches", Date.now(), batchId);
await queue.addBulk(rows.map(row => ({
name: "render-statement",
data: { batchId, ...row },
opts: { jobId: `${batchId}-${row.index}`, attempts: 3, removeOnComplete: 500, removeOnFail: 500 },
})));
await pruneBatches();
return response.status(202).json({ batchId, total: rows.length });
} catch (error) {
if (error.code === "LIMIT_FILE_SIZE") return response.status(413).json({ error: "Keep the CSV under 256 KB." });
return next(error);
}
});
app.get("/api/batches/:batchId", async (request, response) => {
const metadata = await redis.hgetall(`statement:batch:${request.params.batchId}`);
if (!metadata.id) return response.status(404).json({ error: "That batch is no longer available." });
const items = [];
for (let index = 0; index < Number(metadata.total); index += 1) {
const value = await redis.get(`statement:item:${metadata.id}:${index}`);
if (value) items.push(JSON.parse(value));
}
response.json({ ...metadata, total: Number(metadata.total), completed: Number(metadata.completed), failed: Number(metadata.failed), items });
});
app.get("/batches/:batchId/statements/:index.pdf", async (request, response) => {
const key = `statement:pdf:${request.params.batchId}:${request.params.index}`;
const pdf = await redis.getBuffer(key);
if (!pdf) return response.status(404).send("That statement is not ready.");
const item = JSON.parse(await redis.get(`statement:item:${request.params.batchId}:${request.params.index}`));
response.type("application/pdf").attachment(`${item.customer.replace(/[^a-z0-9]+/gi, "-").toLowerCase()}-statement.pdf`).send(pdf);
});
app.get("/batches/:batchId/all.zip", async (request, response) => {
const metadata = await redis.hgetall(`statement:batch:${request.params.batchId}`);
if (!metadata.id || Number(metadata.completed) !== Number(metadata.total)) return response.status(409).send("The batch is not complete yet.");
response.attachment(`statements-${metadata.id.slice(0, 8)}.zip`);
const archive = new ZipArchive({ zlib: { level: 6 } });
archive.on("error", error => response.destroy(error));
archive.pipe(response);
for (let index = 0; index < Number(metadata.total); index += 1) {
const [pdf, itemText] = await Promise.all([
redis.getBuffer(`statement:pdf:${metadata.id}:${index}`),
redis.get(`statement:item:${metadata.id}:${index}`),
]);
const item = JSON.parse(itemText);
archive.append(pdf, { name: `${String(index + 1).padStart(3, "0")}-${item.customer.replace(/[^a-z0-9]+/gi, "-").toLowerCase()}.pdf` });
}
await archive.finalize();
});
app.get("/", (_request, response) => response.sendFile("index.html", { root: __dirname }));
app.use((error, _request, response, _next) => {
console.error(error);
response.status(400).json({ error: error.message || "The batch could not be created." });
});
const port = Number(process.env.PORT || 3000);
const server = app.listen(port, "0.0.0.0", () => console.log(`Statement desk ready on port ${port}`));
process.on("SIGTERM", () => server.close(async () => { await queue.close(); await redis.quit(); process.exit(0); }));
worker/package.json Raw
{
"name": "statement-generator-worker",
"version": "1.0.0",
"private": true,
"engines": { "node": ">=24" },
"scripts": { "start": "node worker.js" },
"dependencies": {
"bullmq": "5.80.9",
"ioredis": "5.11.1",
"pdfkit": "0.19.1"
}
}
worker/worker.js Raw
const { Worker } = require("bullmq");
const IORedis = require("ioredis");
const PDFDocument = require("pdfkit");
const redis = new IORedis(process.env.REDIS_URL, { maxRetriesPerRequest: null });
async function markBatchOnce(batchId, markerKey, field) {
return redis.eval(`
if redis.call("SETNX", KEYS[1], "1") == 1 then
redis.call("HINCRBY", KEYS[2], ARGV[1], 1)
return 1
end
return 0
`, 2, markerKey, `statement:batch:${batchId}`, field);
}
function renderStatement(data) {
return new Promise((resolve, reject) => {
const document = new PDFDocument({ size: "LETTER", margins: { top: 56, left: 56, right: 56, bottom: 56 } });
const chunks = [];
document.on("data", chunk => chunks.push(chunk));
document.on("end", () => resolve(Buffer.concat(chunks)));
document.on("error", reject);
document.fillColor("#31594d").fontSize(12).text("NORTHLINE BILLING", { characterSpacing: 1.5 });
document.moveDown(2).fillColor("#222722").font("Helvetica-Bold").fontSize(28).text("Monthly statement");
document.moveDown(0.4).font("Helvetica").fontSize(11).fillColor("#6a6a63").text(data.period);
document.moveDown(2).strokeColor("#d6d0c5").moveTo(56, document.y).lineTo(556, document.y).stroke();
document.moveDown(2).fillColor("#222722").fontSize(10).text("PREPARED FOR");
document.moveDown(0.5).font("Helvetica-Bold").fontSize(18).text(data.customer);
document.font("Helvetica").fontSize(11).fillColor("#6a6a63").text(data.email);
document.moveDown(3).fillColor("#222722").fontSize(10).text("BALANCE DUE");
document.moveDown(0.5).font("Helvetica-Bold").fontSize(34).text(`$${(data.balanceCents / 100).toFixed(2)}`);
document.moveDown(2).font("Helvetica").fontSize(11).fillColor("#55584f").text("Thank you for your business. Please include your statement number with payment.");
document.moveDown(5).fontSize(9).fillColor("#8a887f").text(`Statement ${data.batchId.slice(0, 8).toUpperCase()}-${String(data.index + 1).padStart(3, "0")}`);
document.end();
});
}
const worker = new Worker("statements", async job => {
const pdf = await renderStatement(job.data);
const pdfKey = `statement:pdf:${job.data.batchId}:${job.data.index}`;
const itemKey = `statement:item:${job.data.batchId}:${job.data.index}`;
const doneKey = `statement:done:${job.data.batchId}:${job.data.index}`;
await redis.set(pdfKey, pdf);
await redis.set(itemKey, JSON.stringify({ index: job.data.index, customer: job.data.customer, email: job.data.email, bytes: pdf.length }));
await redis.sadd(`statement:keys:${job.data.batchId}`, pdfKey, itemKey, doneKey);
await markBatchOnce(job.data.batchId, doneKey, "completed");
console.log(JSON.stringify({ event: "statement.ready", batch: job.data.batchId, index: job.data.index, customer: job.data.customer, bytes: pdf.length }));
return { bytes: pdf.length };
}, { connection: redis, concurrency: 2, lockDuration: 15000 });
worker.on("failed", async (job, error) => {
if (job && job.attemptsMade >= job.opts.attempts) {
const failedKey = `statement:failed:${job.data.batchId}:${job.data.index}`;
await redis.sadd(`statement:keys:${job.data.batchId}`, failedKey);
await markBatchOnce(job.data.batchId, failedKey, "failed");
}
console.error(`Statement job ${job?.id || "unknown"} failed. ${error.message}`);
});
worker.on("ready", () => console.log("Statement worker ready for BullMQ jobs"));
worker.on("error", error => console.error(`Statement worker connection error. ${error.message}`));
async function stop(signal) {
console.log(`Received ${signal}. Returning active jobs to BullMQ.`);
await worker.close();
await redis.quit();
process.exit(0);
}
process.on("SIGTERM", () => void stop("SIGTERM"));
process.on("SIGINT", () => void stop("SIGINT"));
Built from revision 52c7b7f