Build a Bulk Data Export System

npm i bullmq ioredis pg pg-cursor exceljs csv-stringify zod hono @hono/node-server \
  @aws-sdk/client-s3 @aws-sdk/s3-request-presigner
npm i -D typescript tsx @types/node @types/pg @types/pg-cursor

CREATE TABLE export_jobs (
  id uuid PRIMARY KEY, account_id uuid NOT NULL, requested_by uuid NOT NULL,
  format text NOT NULL, resource text NOT NULL, status text NOT NULL,
  total_rows bigint, file_size bigint, s3_key text, error_message text,
  created_at timestamptz NOT NULL DEFAULT now(), completed_at timestamptz
);
CREATE INDEX ON export_jobs (account_id, created_at DESC);

import { z } from "zod";
export const ExportRequest = z.object({
  format: z.enum(["csv", "xlsx"]),
  resource: z.enum(["employees", "payroll", "timesheets"]),
  filters: z
    .object({
      dateFrom: z.iso.datetime().optional(),
      dateTo: z.iso.datetime().optional(),
      departmentIds: z.array(z.uuid()).optional(),
      status: z.string().max(40).optional(),
    })
    .default({}),
  columns: z.array(z.string()).optional(), // checked against the allow-list below
});
export type ExportRequest = z.infer<typeof ExportRequest>;

import pg from "pg";
import Cursor from "pg-cursor";
import ExcelJS from "exceljs";
import { stringify } from "csv-stringify";
import { createWriteStream } from "node:fs";
import { Readable } from "node:stream";
import { pipeline } from "node:stream/promises";
import type { ExportRequest } from "./types";
export const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const BATCH = 5000;
const XLSX_MAX_ROWS = 1_048_575; // one sheet holds 1,048,576 rows including the header
export const RESOURCES = {
  employees: ["id", "full_name", "email", "department_id", "status", "created_at"],
  payroll: ["id", "employee_id", "department_id", "gross", "net", "status", "created_at"],
  timesheets: ["id", "employee_id", "department_id", "hours", "status", "created_at"],
} as const;
function buildQuery(accountId: string, req: ExportRequest) {
  const all: readonly string[] = RESOURCES[req.resource];
  const columns = req.columns?.length ? req.columns : [...all];
  const bad = columns.filter((c) => !all.includes(c));
  if (bad.length) throw new Error(`Unknown columns: ${bad.join(", ")}`);
  const params: unknown[] = [accountId]; // every export is scoped to one tenant
  const where = ["account_id = $1"];
  const { dateFrom, dateTo, departmentIds, status } = req.filters;
  if (dateFrom) where.push(`created_at >= $${params.push(dateFrom)}`);
  if (dateTo) where.push(`created_at <= $${params.push(dateTo)}`);
  if (departmentIds?.length) where.push(`department_id = ANY($${params.push(departmentIds)})`);
  if (status) where.push(`status = $${params.push(status)}`);
  const from = `FROM ${req.resource} WHERE ${where.join(" AND ")}`;
  return {
    columns,
    params,
    count: `SELECT count(*) ${from}`,
    data: `SELECT ${columns.join(", ")} ${from} ORDER BY created_at DESC, id`,
  };
}
// Formula injection: Excel runs cells starting with = + - @ tab or CR. pg returns numeric
// columns as strings, so plain numbers such as "-20.25" are left alone
const NUMBER = /^-?\d+(\.\d+)?$/;
const safe = (v: unknown) =>
  typeof v === "string" && /^[=+\-@\t\r]/.test(v) && !NUMBER.test(v) ? `'${v}` : v;
const sanitize = (row: Record<string, unknown>) =>
  Object.fromEntries(Object.entries(row).map(([k, v]) => [k, safe(v)]));
async function* rows(sql: string, params: unknown[], onBatch: (n: number) => Promise<void>) {
  const client = await pool.connect();
  const cursor = client.query(new Cursor(sql, params)); // server-side cursor, no BEGIN needed
  let done = 0;
  try {
    for (let batch = await cursor.read(BATCH); batch.length; batch = await cursor.read(BATCH)) {
      for (const row of batch) yield sanitize(row);
      done += batch.length;
      await onBatch(done);
    }
  } finally {
    await cursor.close();
    client.release();
  }
}
export async function generate(
  accountId: string, req: ExportRequest, path: string,
  onProgress: (done: number, total: number) => Promise<void>,
) {
  const q = buildQuery(accountId, req);
  const total = Number((await pool.query(q.count, q.params)).rows[0].count);
  if (req.format === "xlsx" && total > XLSX_MAX_ROWS) {
    throw new Error(`Too many rows for one Excel sheet (${total}); export CSV instead`);
  }
  const source = rows(q.data, q.params, (n) => onProgress(n, total));
  if (req.format === "csv") {
    await pipeline(
      Readable.from(source),
      // csv-stringify turns Date into epoch milliseconds by default
      stringify({ header: true, columns: q.columns, cast: { date: (d) => d.toISOString() } }),
      createWriteStream(path),
    );
  } else {
    const wb = new ExcelJS.stream.xlsx.WorkbookWriter({ filename: path });
    const sheet = wb.addWorksheet(req.resource);
    sheet.columns = q.columns.map((key) => ({ header: key.replace(/_/g, " "), key, width: 20 }));
    for await (const row of source) sheet.addRow(row).commit();
    await sheet.commit();
    await wb.commit();
  }
  return total;
}

import { Queue, Worker } from "bullmq";
import { Redis } from "ioredis";
import { randomUUID } from "node:crypto";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { unlink } from "node:fs/promises";
import { generate, pool } from "./processor";
import { upload } from "./storage";
import type { ExportRequest } from "./types";
// BullMQ workers use blocking commands: maxRetriesPerRequest must be null
const connection = new Redis(process.env.REDIS_URL ?? "redis://127.0.0.1:6379", {
  maxRetriesPerRequest: null,
});
export const exportQueue = new Queue("exports", { connection });
type JobData = { jobId: string; accountId: string; request: ExportRequest };
export async function submitExport(accountId: string, userId: string, request: ExportRequest) {
  const jobId = randomUUID();
  await pool.query(
    "INSERT INTO export_jobs (id, account_id, requested_by, format, resource, status) VALUES ($1,$2,$3,$4,$5,'queued')",
    [jobId, accountId, userId, request.format, request.resource]);
  await exportQueue.add("generate", { jobId, accountId, request } satisfies JobData, {
    jobId, attempts: 2, backoff: { type: "fixed", delay: 10_000 },
    removeOnComplete: { age: 86_400 }, removeOnFail: { age: 604_800 } });
  return jobId;
}
export const worker = new Worker<JobData>(
  "exports",
  async (job) => {
    const { jobId, accountId, request } = job.data;
    const path = join(tmpdir(), `export-${jobId}.${request.format}`);
    try {
      await pool.query("UPDATE export_jobs SET status = 'processing' WHERE id = $1", [jobId]);
      const total = await generate(accountId, request, path, (done, total) =>
        job.updateProgress(Math.floor((done / Math.max(total, 1)) * 100))); // stored in Redis
      const key = `exports/${accountId}/${jobId}.${request.format}`;
      const size = await upload(path, key);
      await pool.query(
        `UPDATE export_jobs SET status = 'completed', total_rows = $2, file_size = $3,
           s3_key = $4, completed_at = now() WHERE id = $1`,
        [jobId, total, size, key],
      );
    } finally {
      await unlink(path).catch(() => {});
    }
  },
  { connection, concurrency: 3, limiter: { max: 10, duration: 60_000 } },
);
// Mark the row failed only when the last attempt has failed
worker.on("failed", async (job, err) => {
  if (job && job.attemptsMade >= (job.opts.attempts ?? 1)) {
    await pool.query("UPDATE export_jobs SET status = 'failed', error_message = $2 WHERE id = $1",
      [job.data.jobId, err.message.slice(0, 500)]);
  }
});

import { S3Client, PutObjectCommand, GetObjectCommand } from "@aws-sdk/client-s3";
import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
import { createReadStream } from "node:fs";
import { stat } from "node:fs/promises";
const s3 = new S3Client({ endpoint: process.env.S3_ENDPOINT, forcePathStyle: !!process.env.S3_ENDPOINT });
const Bucket = process.env.S3_BUCKET!;
export async function upload(path: string, Key: string) {
  const { size } = await stat(path);
  await s3.send(new PutObjectCommand({ Bucket, Key, Body: createReadStream(path), ContentLength: size }));
  return size;
}
// Signed on every request, so the link only exists after an authorised GET
export const signedUrl = (Key: string, seconds = 900) =>
  getSignedUrl(s3, new GetObjectCommand({ Bucket, Key }), { expiresIn: seconds });

import { Hono } from "hono";
import { ExportRequest } from "./types";
import { exportQueue, submitExport } from "./queue";
import { pool } from "./processor";
import { signedUrl } from "./storage";
type Env = { Variables: { accountId: string; userId: string } };
export const app = new Hono<Env>();
app.post("/exports", async (c) => {
  const parsed = ExportRequest.safeParse(await c.req.json().catch(() => null));
  if (!parsed.success) return c.json({ error: parsed.error.issues }, 400);
  const accountId = c.get("accountId");
  const { rows } = await pool.query(
    "SELECT count(*)::int AS active FROM export_jobs WHERE account_id = $1 AND status IN ('queued','processing')", [accountId]);
  if (rows[0].active >= 5) return c.json({ error: "Maximum 5 concurrent exports" }, 429);
  const id = await submitExport(accountId, c.get("userId"), parsed.data);
  return c.json({ id, status: "queued", pollUrl: `/exports/${id}` }, 202);
});
app.get("/exports/:id", async (c) => {
  const { rows } = await pool.query(
    `SELECT id, format, resource, status, total_rows, file_size, s3_key, error_message, created_at, completed_at
     FROM export_jobs WHERE id = $1 AND account_id = $2`, // the tenant check applies to every read
    [c.req.param("id"), c.get("accountId")],
  );
  if (!rows.length) return c.json({ error: "Export not found" }, 404);
  const { s3_key, ...job } = rows[0];
  if (job.status === "processing") {
    const live = await exportQueue.getJob(job.id);
    return c.json({ ...job, progress: live?.progress ?? 0 });
  }
  if (job.status === "completed") {
    return c.json({ ...job, progress: 100, downloadUrl: await signedUrl(s3_key) });
  }
  return c.json(job);
});