Backfilling moderation over posts you already have: a Node bulk job

Classifying an existing table of posts and comments is a data-migration problem wearing an AI hat. Keyset cursors, idempotent writeback, and an auditable export.

Moderating content you’re about to receive and moderating four years of content you already have are different engineering problems. The second one is a data migration that happens to call a model: you need a cursor that survives a restart, a writeback that’s safe to run twice, and an artefact the trust-and-safety team can audit afterwards. Infrai’s batch endpoint handles the inference half in one call per chunk, but the parts that decide whether your backfill finishes are all on your side of the wire.

This walks the whole job in Node 22 against a Postgres table of posts and comments. For the mechanics of the batch API itself — chunk sizing, poll states, retries — see the complete batch job walkthrough; here we assume it works and concentrate on the backlog.

First: you’re classifying with a chat model, not a moderation endpoint

Worth checking before you design anything, because the obvious route is a dead end today:

export INFRAI_API_KEY="your_infrai_api_key"

curl -sS -X POST "https://api.infrai.cc/v1/moderations" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"model":"auto","input":"I will hurt you"}'

{
  "error": {
    "message": "ai-runtime vendor not configured for ai.moderation — hydrate an AI vendor key",
    "type": "api_error",
    "code": "VENDOR_NOT_CONFIGURED",
    "param": null
  }
}

No moderation vendor is keyed on the shared pool, so the OpenAI-compatible moderations route answers with that rather than category scores. Classification through a cheap chat model is what works, and it has one advantage for a backfill: you define the taxonomy, so historical content gets judged against the policy you have now instead of a vendor’s generic categories.

If you’re already an OpenAI customer, their moderation endpoint is free, purpose-built and better calibrated than a prompt. For a one-off backfill of your own archive, use it and skip this article — that’s the genuinely cheaper answer and we’d rather say so.

The checkpoint table is the job

Everything else is replaceable. This isn’t.

CREATE TABLE moderation_backfill (
  content_id   bigint PRIMARY KEY,
  content_kind text        NOT NULL,
  label        text,
  confidence   numeric(4,3),
  model        text,
  batch_id     text,
  cost_usd     numeric(12,8),
  decided_at   timestamptz
);

CREATE INDEX ON moderation_backfill (decided_at) WHERE decided_at IS NULL;

Two properties matter. The primary key means a rerun over the same rows is an upsert rather than a duplicate, and the partial index means “what’s left to do” is a fast query no matter how far in you are. Seed it once from the tables you’re backfilling, then never read the source tables for progress again.

Cursor by primary key, not by OFFSET. A 900,000-row backfill with an offset scan gets quadratically slower and you’ll blame the model:

SELECT p.id, p.body
FROM posts p
JOIN moderation_backfill m ON m.content_id = p.id
WHERE m.decided_at IS NULL
  AND p.id > $1
ORDER BY p.id
LIMIT 100;

The chunk runner

Each chunk is: read 100 rows, ask for a label and a confidence in JSON, write the answers back, advance. Persist batch_id at submit time — GET /v1/ai/batch/list came back empty for us immediately after a successful store: true submit, so it isn’t a recovery path:

curl -sS "https://api.infrai.cc/v1/ai/batch/list" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

{ "ok": true, "data": { "batches": [], "total_count": 0, "next_cursor": null } }

Here’s the runner. Note that request_index is the only join key between what you sent and what came back, so the array order is load-bearing — never filter or sort the row array between building it and reading the results.

import pg from "pg";

const KEY = process.env.INFRAI_API_KEY;
if (!KEY) throw new Error("INFRAI_API_KEY is not set");
const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

const POLICY = 'Classify the text. Reply with JSON only: {"label":"ok|spam|harassment|sexual|violence","confidence":0.0-1.0}';

function buildPayload(rows) {
  return JSON.stringify({
    requests: rows.map((r) => ({
      model: "glm-4-air",
      max_tokens: 80,
      response_format: { type: "json_object" },
      messages: [
        { role: "system", content: POLICY },
        { role: "user", content: r.body.slice(0, 4000) },
      ],
    })),
    batch_timeout: 1800,
    metadata: { job: "moderation-backfill", low: String(rows[0].id) },
    store: true,
  });
}

function parseDecision(raw) {
  try {
    const d = JSON.parse(raw);
    const label = String(d.label ?? "");
    const confidence = Number(d.confidence);
    if (!label || Number.isNaN(confidence)) return null;
    return { label, confidence: Math.min(1, Math.max(0, confidence)) };
  } catch {
    return null;
  }
}

export async function runChunk(rows) {
  const submit = await fetch("https://api.infrai.cc/v1/ai/batch/submit", {
    method: "POST",
    headers: { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" },
    body: buildPayload(rows),
  });
  if (!submit.ok) throw new Error(`submit failed: ${submit.status} ${await submit.text()}`);
  const { data: job } = await submit.json();

  await pool.query(
    "UPDATE moderation_backfill SET batch_id = $1 WHERE content_id = ANY($2::bigint[])",
    [job.batch_id, rows.map((r) => r.id)],
  );

  const results = await fetch(`https://api.infrai.cc/v1/ai/batch/results/${job.batch_id}`, {
    headers: { Authorization: `Bearer ${KEY}` },
  });
  if (!results.ok) throw new Error(`results failed: ${results.status}`);
  const { data } = await results.json();

  let written = 0;
  let unparseable = 0;
  for (const item of data.items) {
    const row = rows[item.request_index];
    if (!item.ok) continue;
    const decision = parseDecision(item.result.content);
    if (!decision) { unparseable++; continue; }
    await pool.query(
      `UPDATE moderation_backfill
          SET label = $1, confidence = $2, model = $3, cost_usd = $4, decided_at = now()
        WHERE content_id = $5`,
      [decision.label, decision.confidence, item.result.model, item.cost_usd ?? 0, row.id],
    );
    written++;
  }
  return { batchId: job.batch_id, written, unparseable, left: data.items.length - written - unparseable };
}

That parseDecision guard is not defensive padding. response_format: { type: "json_object" } is honoured on the models we tried, but a stricter json_schema is not enforced on the cheap ones — you’ll get valid JSON with a label your enum never contained, or a fenced block wrapping it. Treat every model answer as untrusted input and count the failures; if unparseable climbs above a percent or two, the prompt is wrong, not the parser.

Rows that fail stay decided_at IS NULL and get picked up by the next pass. That’s the whole retry strategy, and it’s better than an in-memory retry queue because it survives the process dying at 3 a.m.

Export, because someone will ask

A backfill that changes 900,000 moderation states will be questioned. Keep the raw model output as a file, not just the derived labels:

curl -sS -X POST "https://api.infrai.cc/v1/ai/batch/export/{batch_id}" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"format": "jsonl"}'

{
  "ok": true,
  "data": {
    "batch_id": "batch_8393d98606fd3e4f47090d7f",
    "format": "jsonl",
    "content": "{\"request_index\": 0, \"ok\": true, \"result\": {\"content\": \"{\\\"label\\\":\\\"ok\\\",\\\"confidence\\\":0.97}\", \"model\": \"glm-4-air\"}, \"error\": null}\n",
    "total_count": 2
  }
}

format also accepts csv, which is what the people asking usually want. Export is free, so do it per chunk and stream the lines into object storage on the same key rather than holding a 900,000-line string in memory.

What a real backfill costs and how long it takes

The submit call blocks while its rows run, so throughput is a concurrency question rather than a pricing one. Run several chunks at once with a small worker pool, keep each chunk at 100 rows so a client timeout costs you one chunk and not one hour, and read your actual spend from the per-row cost_usd you’re already storing — that’s a SELECT sum(cost_usd) away rather than a reconciliation.

Confirm the total against the platform’s own ledger when you’re done:

curl -sS "https://api.infrai.cc/v1/account/usage" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

The breakdown array has an ai.chat row with cost and calls; divide one by the other for your true effective rate across the whole run, verified on the day you ran it. These rates keep drifting down, so a backfill quoted three months ago is probably cheaper now.

Limitations, and when to use something else

Text only — this path doesn’t support image or video rows, so a platform with user-uploaded media needs a second tool for that surface. There’s no policy-tuned taxonomy either, which means calibration against a labelled sample is your job, permanently. And every row is routed as a chat request, so don’t try to sneak an embedding or an image call into the array.

For continuous moderation of new content, a backfill runner is the wrong shape entirely — you want a tiered live pipeline, which we cover in the high-volume moderation guide. Anthropic publish a content-moderation guide for Claude that’s the best free writeup of the prompt-engineering side, and for regulated categories at scale a trust-and-safety vendor with human review tooling will beat a DIY classifier on both quality and audit trail.

What keeps a backfill here is everything around the classifier: the export lands in object storage, the review items go on a queue, the failure at row 412,000 shows up in error tracking, and the whole thing bills to one account you can attribute per tenant. One credential, one invoice, one usage view — for a job that runs for five days unattended, that’s worth more than a marginally better label.

References

• Infrai AI runtime API reference
• VENDOR_NOT_CONFIGURED error reference
• OpenAI moderation guide
• OpenAI batch API guide