Support Catalog Backfill: Moderate Existing Posts and Comments in a Node.js Bulk Job

#node #llm #observability

Per-tenant cost visibility changes the design: don't begin with parallel API calls; begin with a durable ledger that ties every classification result and usage record to a tenant, policy version, and source item. For a customer-support catalog backfill, the practical choice is a bounded Node.js worker that reads existing posts and comments, classifies them through a replaceable adapter, checkpoints each result, and exports tenant-scoped JSONL.

Short answer: make the ledger the product of the job and the LLM call one restartable step inside it.

That ordering matters when support conversations contain messy product descriptions such as “the small blue charger for the old tablet.” The moderation label decides whether the text is safe to reuse; the enrichment labels connect it to a catalog candidate. Operations still need to answer a less glamorous question: which tenant consumed the tokens?

Make tenant cost visible before optimizing it

Token totals belong beside decisions, not in an unrelated monthly dashboard. Record normalized input and output token counts on every completed row, then aggregate by tenantId, policyVersion, and time window. If the API reports different usage units, preserve the raw usage payload in restricted telemetry and map it explicitly; don't pretend unlike units are interchangeable.

Start there.

Three signals are enough for the first useful view:

Signal	Group by	Operational question
Completed items	tenant, policy version	Is the backfill moving?
Input and output tokens	tenant, model	Where is consumption occurring?
Review and block counts	tenant, content kind	Did the decision mix shift?

Cost in currency should be derived from a versioned rate configuration, not baked into historical rows. Store usage and the model identifier, then apply the applicable rate when producing a report. This keeps a rate change from rewriting what the runtime actually observed. It also lets finance reproduce an invoice-period view while engineering inspects tokens per catalog item.

Watch cardinality. Tenant IDs are useful dimensions in logs and ledger queries, but a metrics backend can become expensive or hard to operate when every item ID becomes a label. Put itemId and the stable key in structured logs or traces. Keep metrics aggregated by bounded dimensions such as policy version, content kind, and worker outcome; whether tenant ID is acceptable as a metric label depends on tenant count and the limits of the telemetry system. I'm not sure there is one universal cutoff — measure series growth in the backend you use.

One more warning: a falling token average isn't automatically good news. It may mean descriptions became cleaner. It may also mean truncation removed the evidence needed for a correct moderation decision. Pair consumption with review rate and a labeled evaluation sample.

Replace a request loop with a ledger

The tempting before picture is simple: query every old comment, call a classification API, then write one large results file. It works until process exit code 137, a deployment, or a rate limit lands after 38,000 items. At that point, an output file answers neither “what was committed?” nor “what may be sent again?”

The after picture has four named stages — source, ledger, classifier, export. The source yields immutable item IDs. The ledger records a terminal result for each (tenantId, itemId, policyVersion) key. The classifier adapter turns one chunk into normalized decisions and usage. The exporter reads committed rows rather than live API responses. Each arrow can stop independently, and every operational graph can use the same tenant dimension.

Keep moderation and catalog enrichment distinct in the result even if one prompt produces both. moderationLabel: "allow" and catalogTags: ["charger", "tablet"] have different consumers, retention rules, and review paths. A single vague classification field makes later policy changes painful.

This is the crisp mental model: requests are temporary; decisions are durable.

How should a Node.js bulk job classify existing posts and comments?

Use a small concurrency limit, stable item keys, and an adapter that normalizes the API response. The example below expects newline-delimited JSON as input and output because an interrupted append leaves earlier lines readable. It deliberately accepts the classifier URL and model identifier through configuration; the endpoint contract and usage fields must be mapped to the API you actually operate.

import { appendFile, readFile } from "node:fs/promises";
import { createHash } from "node:crypto";

type SourceItem = {
  tenantId: string;
  itemId: string;
  kind: "post" | "comment";
  text: string;
};

type Decision = {
  moderationLabel: "allow" | "review" | "block";
  catalogTags: string[];
};

type Usage = {
  inputTokens: number;
  outputTokens: number;
};

type LedgerRow = SourceItem & {
  key: string;
  policyVersion: string;
  decision: Decision;
  usage: Usage;
  completedAt: string;
};

type ClassifierResponse = {
  decision: Decision;
  usage: Usage;
};

const policyVersion = "catalog-moderation-v3";
const outputPath = "moderation-results.jsonl";
const classifierUrl = process.env.CLASSIFIER_URL;
const classifierModel = process.env.CLASSIFIER_MODEL;

if (!classifierUrl || !classifierModel) {
  throw new Error("CLASSIFIER_URL and CLASSIFIER_MODEL are required");
}

function stableKey(item: SourceItem): string {
  return createHash("sha256")
    .update(`${item.tenantId}\0${item.itemId}\0${policyVersion}`)
    .digest("hex");
}

async function classify(item: SourceItem): Promise<ClassifierResponse> {
  const response = await fetch(classifierUrl, {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({
      model: classifierModel,
      policyVersion,
      input: { kind: item.kind, text: item.text },
      outputSchema: {
        moderationLabel: ["allow", "review", "block"],
        catalogTags: "string[]",
      },
    }),
  });

  if (response.status === 429) {
    throw new Error("RATE_LIMITED");
  }
  if (!response.ok) {
    throw new Error(`CLASSIFIER_REJECTED_${response.status}`);
  }

  return (await response.json()) as ClassifierResponse;
}

async function loadCompletedKeys(path: string): Promise<Set<string>> {
  try {
    const data = await readFile(path, "utf8");
    return new Set(
      data
        .split("\n")
        .filter(Boolean)
        .map((line) => (JSON.parse(line) as LedgerRow).key),
    );
  } catch (error) {
    if ((error as NodeJS.ErrnoException).code === "ENOENT") return new Set();
    throw error;
  }
}

async function run(items: SourceItem[]): Promise<void> {
  const completed = await loadCompletedKeys(outputPath);

  for (const item of items) {
    const key = stableKey(item);
    if (completed.has(key)) continue;

    const result = await classify(item);
    const row: LedgerRow = {
      ...item,
      key,
      policyVersion,
      decision: result.decision,
      usage: result.usage,
      completedAt: new Date().toISOString(),
    };

    await appendFile(outputPath, `${JSON.stringify(row)}\n`, "utf8");
    completed.add(key);
  }
}

const input = await readFile("catalog-support-items.json", "utf8");
await run(JSON.parse(input) as SourceItem[]);

For a production database, enforce the stable key with a unique constraint and commit the decision plus usage in one transaction. The file example has a narrower promise: it makes the data shape, checkpoint, and tenant attribution visible without hiding the important parts behind a queue library. It runs serially on purpose. Add a worker pool only after measuring the classifier's documented rate limits, then cap both global concurrency and per-tenant concurrency so one large catalog cannot starve the others.

Don't blindly retry the POST. RFC 9110 defines idempotent methods and explains why clients can automatically retry idempotent requests after a communication failure; POST is not idempotent by definition. If the classification API documents an idempotency mechanism, send the stable key using that documented contract. Otherwise, record an attempt before dispatch, reconcile ambiguous outcomes, and retry only according to the provider's semantics. A local completed-key check prevents ordinary replay after success, but it cannot prove that an interrupted remote request did or did not finish.

How can you export LLM results without leaking tenant data?

Treat export as a query over committed ledger rows. Filter by tenant before serialization, write to a tenant-specific destination, and include the policy version so downstream support tools can explain why a comment was allowed or held. An export row should contain the source key, decision, catalog tags, usage, and completion time; raw text should be included only when the downstream workflow needs it and the retention policy permits it.

The catch is that JSONL is not suitable when analysts need frequent joins, fine-grained access controls, or mutable review status. Keep the ledger in Postgres in that case and generate immutable export objects from a repeatable query. Stick with object storage when exports are append-only handoffs to a warehouse or audit process. A queue helps distribute classification work, but it does not replace the ledger: delivery state and business state answer different questions.

Before releasing a tenant export, test the boundary with two fixtures whose item IDs deliberately collide across tenants. The stable key must remain different, each export must contain only its requested tenant, and aggregate usage must reconcile with the included rows. This tiny test catches a dangerous shortcut: keying solely by itemId because it looked globally unique in the first dataset.

What about prompt drift and human review?

Version the complete decision contract: instructions, label definitions, output schema, and model identifier. Prompt engineering guidance can help shape clear instructions, but a production change still needs a representative labeled set. Run the old and candidate policy over that set, compare label transitions, and inspect cases that move into or out of block. Then start the catalog backfill under a new policyVersion; never silently mix changed rules into an existing run.

Human review is not an exception path. Route review decisions to a queue with the source reference and policy version, capture the final disposition separately, and use those adjudicated cases to improve the next evaluation set. Don't overwrite the model decision. Preserving both answers shows where the policy or prompt needs work and keeps the original export auditable.

There is a real limitation here: LLM classification is not suitable when a rule must be perfectly deterministic, cheaply expressed as code, or decided with evidence the model cannot access. Use deterministic filters for exact allowlists, file types, and known identifiers. Use a human decision for high-impact ambiguous cases. The runtime should orchestrate those layers, not ask one probabilistic call to carry the entire policy.

Ship the smallest observable loop first: ten fixtures, one tenant-scoped ledger, one export, and a usage reconciliation check. Then increase concurrency. Fast is useful only after restart, isolation, and accounting are boring.