Build Your Own PDF API, Part 3: Async Jobs, Storage and Webhooks With Plain Postgres

client ──POST /jobs──▶ API ──INSERT──▶ pdf_jobs table ◀──claim── worker(s)
                         │                                           │
client ◀─202 + job id────┘                     render → upload to S3 → webhook
client ──GET /jobs/:id──▶ API ──▶ status + signed download URL

CREATE TABLE IF NOT EXISTS pdf_jobs (
  id           uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  status       text NOT NULL DEFAULT 'queued'
               CHECK (status IN ('queued', 'running', 'done', 'failed')),
  template     text NOT NULL,
  data         jsonb NOT NULL DEFAULT '{}',
  webhook_url  text,
  attempts     int  NOT NULL DEFAULT 0,
  max_attempts int  NOT NULL DEFAULT 3,
  run_after    timestamptz NOT NULL DEFAULT now(),
  locked_at    timestamptz,
  object_key   text,
  error        text,
  created_at   timestamptz NOT NULL DEFAULT now(),
  finished_at  timestamptz
);
 
-- The worker's claim query only ever looks at queued jobs that are due.
CREATE INDEX IF NOT EXISTS pdf_jobs_due ON pdf_jobs (run_after) WHERE status = 'queued';

npm install pg @aws-sdk/client-s3 @aws-sdk/s3-request-presigner

// storage.js — works with AWS S3, Cloudflare R2, MinIO, or any S3-compatible store
const { S3Client, PutObjectCommand, GetObjectCommand } = require("@aws-sdk/client-s3");
const { getSignedUrl } = require("@aws-sdk/s3-request-presigner");
 
const s3 = new S3Client({
  region: process.env.S3_REGION || "auto",
  endpoint: process.env.S3_ENDPOINT, // e.g. https://<account>.r2.cloudflarestorage.com
  forcePathStyle: true,
  credentials: {
    accessKeyId: process.env.S3_ACCESS_KEY_ID,
    secretAccessKey: process.env.S3_SECRET_ACCESS_KEY,
  },
});
const Bucket = process.env.S3_BUCKET;
 
async function putPdf(key, buffer) {
  await s3.send(new PutObjectCommand({ Bucket, Key: key, Body: buffer, ContentType: "application/pdf" }));
}
 
// Hand out short-lived links instead of making the bucket public.
function signedUrl(key, expiresIn = 3600) {
  return getSignedUrl(s3, new GetObjectCommand({ Bucket, Key: key }), { expiresIn });
}
 
module.exports = { putPdf, signedUrl };

// api.js — accept jobs fast, let workers do the slow part
const express = require("express");
const { Pool } = require("pg");
const { signedUrl } = require("./storage");
 
const db = new Pool({ connectionString: process.env.DATABASE_URL });
// Without this, one dropped idle connection (DB restart, failover) crashes the process.
db.on("error", (err) => console.error("idle pg client error:", err.message));
const app = express();
app.use(express.json({ limit: "5mb" }));
 
app.post("/jobs", async (req, res) => {
  const { template, data = {}, webhookUrl = null } = req.body;
  if (typeof template !== "string" || !template) {
    return res.status(400).json({ error: "template is required" });
  }
  const { rows } = await db.query(
    `INSERT INTO pdf_jobs (template, data, webhook_url) VALUES ($1, $2, $3) RETURNING id, status`,
    [template, data, webhookUrl]
  );
  res.status(202).json(rows[0]); // 202 Accepted: "we have it, check back later"
});
 
app.get("/jobs/:id", async (req, res) => {
  const { rows } = await db.query(
    `SELECT id, status, attempts, error, object_key, created_at, finished_at FROM pdf_jobs WHERE id = $1`,
    [req.params.id]
  );
  if (!rows[0]) return res.status(404).json({ error: "not found" });
  const { object_key, ...job } = rows[0];
  if (job.status === "done") job.url = await signedUrl(object_key);
  res.json(job);
});
 
module.exports = { app, db };
if (require.main === module) app.listen(process.env.PORT || 3000);

// worker.js — claim a job, render it, store it, notify
const crypto = require("crypto");
const puppeteer = require("puppeteer");
const { Pool } = require("pg");
const { renderHtml, renderPdf } = require("./render");
const { putPdf } = require("./storage");
 
const db = new Pool({ connectionString: process.env.DATABASE_URL });
// Without this, one dropped idle connection (DB restart, failover) crashes the process.
db.on("error", (err) => console.error("idle pg client error:", err.message));
const STALE_AFTER = "5 minutes"; // a job "running" longer than this is presumed dead
 
// Atomically take one due job. SKIP LOCKED means two workers never get the same row,
// and neither waits on the other.
async function claimJob() {
  const { rows } = await db.query(`
    UPDATE pdf_jobs SET status = 'running', locked_at = now(), attempts = attempts + 1
    WHERE id = (
      SELECT id FROM pdf_jobs
      WHERE status = 'queued' AND run_after <= now()
      ORDER BY run_after
      FOR UPDATE SKIP LOCKED
      LIMIT 1
    )
    RETURNING *`);
  return rows[0];
}
 
// If a worker crashes mid-job, its row stays 'running' forever. Put it back.
async function reapStaleJobs() {
  await db.query(`
    UPDATE pdf_jobs SET status = 'queued', locked_at = NULL
    WHERE status = 'running' AND locked_at < now() - $1::interval`, [STALE_AFTER]);
}
 
class PermanentError extends Error {} // retrying won't help (bad template, bad data)
 
async function processJob(browser, job) {
  let html;
  try {
    html = renderHtml(job.template, job.data);
  } catch (err) {
    throw new PermanentError(`template error: ${err.message}`);
  }
  const pdf = await renderPdf(browser, html);
  const key = `pdfs/${job.id}.pdf`;
  await putPdf(key, pdf);
  return key;
}
 
async function finish(job, fields) {
  await db.query(
    `UPDATE pdf_jobs SET status = $2, object_key = $3, error = $4, finished_at = now(), locked_at = NULL
     WHERE id = $1`,
    [job.id, fields.status, fields.objectKey || null, fields.error || null]
  );
  if (job.webhook_url) await notify(job.webhook_url, { id: job.id, status: fields.status, error: fields.error || null });
}
 
async function handleFailure(job, err) {
  const retryable = !(err instanceof PermanentError) && job.attempts < job.max_attempts;
  if (!retryable) return finish(job, { status: "failed", error: err.message });
  const delaySec = 5 * 2 ** (job.attempts - 1); // 5s, 10s, 20s...
  await db.query(
    `UPDATE pdf_jobs SET status = 'queued', locked_at = NULL, error = $2,
       run_after = now() + make_interval(secs => $3) WHERE id = $1`,
    [job.id, err.message, delaySec]
  );
}
 
// Sign webhook bodies so receivers can verify they came from you.
async function notify(url, payload) {
  const body = JSON.stringify(payload);
  const sig = crypto.createHmac("sha256", process.env.WEBHOOK_SECRET || "dev").update(body).digest("hex");
  try {
    await fetch(url, {
      method: "POST",
      headers: { "content-type": "application/json", "x-signature": sig },
      body,
      signal: AbortSignal.timeout(10_000),
    });
  } catch (err) {
    console.error(`webhook to ${url} failed: ${err.message}`); // don't fail the job over it
  }
}
 
async function main({ concurrency = 2, pollMs = 1000 } = {}) {
  const browser = await puppeteer.launch({ args: ["--no-sandbox", "--disable-dev-shm-usage"] });
  let stopping = false;
  process.on("SIGTERM", () => (stopping = true)); // finish current jobs, then exit
 
  setInterval(() => reapStaleJobs().catch(console.error), 60_000).unref();
 
  const loop = async () => {
    while (!stopping) {
      const job = await claimJob();
      if (!job) { await new Promise((r) => setTimeout(r, pollMs)); continue; }
      try {
        const objectKey = await processJob(browser, job);
        await finish(job, { status: "done", objectKey });
      } catch (err) {
        await handleFailure(job, err);
      }
    }
  };
  await Promise.all(Array.from({ length: concurrency }, loop));
  await browser.close();
  await db.end();
}
 
module.exports = { main, claimJob, reapStaleJobs };
if (require.main === module) main();

SELECT id FROM pdf_jobs WHERE status = 'queued' AND run_after <= now()
ORDER BY run_after FOR UPDATE SKIP LOCKED LIMIT 1

const crypto = require("crypto");
 
function verifyWebhook(rawBody, signature, secret) {
  const expected = crypto.createHmac("sha256", secret).update(rawBody).digest("hex");
  return signature.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected));
}

psql "$DATABASE_URL" -f schema.sql
DATABASE_URL=... S3_ENDPOINT=... S3_BUCKET=... S3_ACCESS_KEY_ID=... S3_SECRET_ACCESS_KEY=... node api.js
DATABASE_URL=... S3_ENDPOINT=... S3_BUCKET=... S3_ACCESS_KEY_ID=... S3_SECRET_ACCESS_KEY=... node worker.js

curl -X POST localhost:3000/jobs -H 'content-type: application/json' \
  -d '{"template":"<h1>Hi {{name}}</h1>","data":{"name":"Ana"}}'
# {"id":"6f1c...","status":"queued"}
 
curl localhost:3000/jobs/6f1c...
# {"id":"6f1c...","status":"done","attempts":1,"url":"https://...signed..."}