7-Day Content Moderation Cost Ledger Explained (Chat Completions with JSON Schema)

Short answer: put text and image classification behind one schema-checked moderation job, charge usage to the tenant recorded on that immutable job, and make every retry reuse the same idempotency key. For an edtech review queue, this is the least complex design that keeps unsafe uploads away from reviewers while making each tenant's consumption explainable.

The model call is only one step. The production unit is a state transition: accepted, dispatched, classified, or sent to human review. If those states and their cost evidence cannot be replayed, a valid JSON response won't save the system during an incident.

What an incident changes about the design

I've been paged by missed jobs and duplicate deliveries. The tempting response is to tune a timeout, but the lasting lesson is narrower: delivery is an attempt, not proof of completion. A moderation worker must be able to receive the same report twice without creating two review decisions, and an operator must be able to find accepted reports that never reached a terminal state.

Consider a course platform where schools are separate tenants. A learner flags a discussion post containing text and one screenshot. The gateway assigns an opaque report ID, stores the tenant ID and content hashes, then enqueues the report. It does not wait for classification. A worker claims the job, sends the text and image to an OpenAI-compatible chat completions endpoint, validates the returned object, and records both the decision and usage evidence in one commit. Only then does it acknowledge the queue message. If the acknowledgement is lost, the next delivery finds the completed report and exits. If dispatch never happens, a scheduled sweep finds the nonterminal report by age and re-enqueues it.

That's the invariant: one report ID can produce many attempts but only one effective moderation decision.

A 7-day attribution window is a reporting boundary, not a retention recommendation. Keep a ledger row per attempt with tenant ID, report ID, model identifier, input modality, provider request ID when one exists, token or unit counts returned by the API, timestamps, and an outcome category. Never estimate one tenant's share by dividing a global invoice by request count; a long post with an image and a two-word comment are not equivalent work. I'm not sure every compatible endpoint exposes enough usage detail for exact image allocation. A capability probe against the chosen endpoint, followed by invoice reconciliation, resolves that uncertainty.

How should a content moderation safety check handle text, images, and JSON schema?

Treat the schema as the internal contract, not as evidence that the classifier is correct. The result needs a stable decision enum, category findings, a reason suitable for a reviewer, and a schema version. Keep provider-specific fields outside that contract. In a Node.js gateway, this can remain an ordinary internal HTTP boundary; the worker below is Go because the concurrency and cancellation path are easier to see without framework code. The wire format is plain JSON, so the gateway language does not own the decision model.

A useful response shape is deliberately small:

{
  "schema_version": "1",
  "decision": "review",
  "categories": ["harassment"],
  "reason": "Targeted insulting language requires a human decision."
}

Use three decisions: allow, review, and block. The middle state matters. For ambiguous classroom speech, forcing a binary model answer turns uncertainty into an automated policy decision. Validate enum values, reject extra properties, cap the reason length, and route invalid output to human review rather than guessing what the model meant.

Text and images should share a report ID but retain separate content hashes. That permits a text-only retry when image retrieval failed before dispatch, avoids storing raw classroom content in logs, and lets an auditor distinguish a changed upload from a duplicate message. The classifier input should contain the policy version as well. Otherwise, replaying last week's report under this week's rules can silently produce a different result with no explanation.

Streaming does not improve this decision path. Server-Sent Events are a one-way server-to-client stream, useful when a browser needs progressive updates, but moderation should commit one validated object. Partial tokens are not a decision and must never unlock content.

The preventative code path

The critical code sits around the model call. It creates a deterministic attempt key, applies a deadline, requires a successful response, rejects unknown JSON fields, checks the schema version and enum, and leaves acknowledgment to the caller only after durable commit. No SDK is required for an OpenAI-compatible API; ordinary HTTP also makes the exact request and response visible in traces.

package moderation

import (
    "bytes"
    "context"
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "errors"
    "fmt"
    "net/http"
    "time"
)

type Report struct {
    ID        string
    TenantID  string
    PolicyVer string
    Text      string
    ImageURL  string
}

type Decision struct {
    SchemaVersion string   `json:"schema_version"`
    Decision      string   `json:"decision"`
    Categories    []string `json:"categories"`
    Reason        string   `json:"reason"`
}

type Usage struct {
    InputTokens  int `json:"prompt_tokens"`
    OutputTokens int `json:"completion_tokens"`
}

type completionResponse struct {
    Choices []struct {
        Message struct {
            Content string `json:"content"`
        } `json:"message"`
    } `json:"choices"`
    Usage Usage `json:"usage"`
}

type Ledger interface {
    CommitOnce(ctx context.Context, attemptKey string, report Report, decision Decision, usage Usage) error
}

func Classify(ctx context.Context, client *http.Client, endpoint, apiKey, model string, report Report, ledger Ledger) error {
    sum := sha256.Sum256([]byte(report.TenantID + "\x00" + report.ID + "\x00" + report.PolicyVer))
    attemptKey := hex.EncodeToString(sum[:])

    payload := map[string]any{
        "model": model,
        "messages": []map[string]any{{
            "role": "user",
            "content": []map[string]any{
                {"type": "text", "text": report.Text},
                {"type": "image_url", "image_url": map[string]string{"url": report.ImageURL}},
            },
        }},
        "response_format": map[string]any{
            "type": "json_schema",
            "json_schema": map[string]any{
                "name": "moderation_decision",
                "strict": true,
                "schema": map[string]any{
                    "type": "object",
                    "additionalProperties": false,
                    "required": []string{"schema_version", "decision", "categories", "reason"},
                    "properties": map[string]any{
                        "schema_version": map[string]any{"type": "string", "const": "1"},
                        "decision": map[string]any{"type": "string", "enum": []string{"allow", "review", "block"}},
                        "categories": map[string]any{"type": "array", "items": map[string]string{"type": "string"}},
                        "reason": map[string]any{"type": "string", "maxLength": 500},
                    },
                },
            },
        },
    }

    body, err := json.Marshal(payload)
    if err != nil {
        return fmt.Errorf("encode request: %w", err)
    }

    callCtx, cancel := context.WithTimeout(ctx, 20*time.Second)
    defer cancel()
    req, err := http.NewRequestWithContext(callCtx, http.MethodPost, endpoint, bytes.NewReader(body))
    if err != nil {
        return fmt.Errorf("build request: %w", err)
    }
    req.Header.Set("Authorization", "Bearer "+apiKey)
    req.Header.Set("Content-Type", "application/json")
    req.Header.Set("Idempotency-Key", attemptKey)

    res, err := client.Do(req)
    if err != nil {
        return fmt.Errorf("call classifier: %w", err)
    }
    defer res.Body.Close()
    if res.StatusCode != http.StatusOK {
        return fmt.Errorf("classifier status: %d", res.StatusCode)
    }

    var wire completionResponse
    decoder := json.NewDecoder(res.Body)
    decoder.DisallowUnknownFields()
    if err := decoder.Decode(&wire); err != nil {
        return fmt.Errorf("decode response: %w", err)
    }
    if len(wire.Choices) != 1 {
        return errors.New("expected one completion choice")
    }

    var decision Decision
    decisionDecoder := json.NewDecoder(bytes.NewBufferString(wire.Choices[0].Message.Content))
    decisionDecoder.DisallowUnknownFields()
    if err := decisionDecoder.Decode(&decision); err != nil {
        return fmt.Errorf("decode decision: %w", err)
    }
    if decision.SchemaVersion != "1" || !validDecision(decision.Decision) {
        return errors.New("decision violates moderation contract")
    }
    return ledger.CommitOnce(ctx, attemptKey, report, decision, wire.Usage)
}

func validDecision(value string) bool {
    return value == "allow" || value == "review" || value == "block"
}

The endpoint is configuration because compatible APIs differ in base URL and deployment naming. Before deployment, run a capability test with one text-only fixture, one image fixture, one schema refusal case, and a forced deadline. Do not put real student content in those fixtures. A 429 should leave the job retryable with bounded backoff; a malformed model result should become review, with the raw response protected under the same access rules as the original report. Keep it boring.

One detail deserves a code review comment: CommitOnce needs a unique constraint on the attempt key or report-and-policy tuple. An in-memory duplicate check is a race. Two workers can both observe absence, both call the model, and both write unless the datastore arbitrates the final transition. Duplicate calls may still consume capacity, but they cannot create conflicting effective decisions.

Operating the queue and the tenant ledger

Alert on user-visible invariants, not raw worker activity. The useful signals are oldest unclassified report age, count of reports outside their review objective, retry attempts by outcome, schema rejection count, and duplicate commit count. Break usage down by tenant, model, modality, and policy version. Then reconcile the sum against the provider's billing export on a fixed cadence; drift is a signal that a response omitted usage, an attempt escaped logging, or the provider accounts for a unit differently.

A runbook should answer four questions in order: Are new reports being accepted? Are queued reports advancing? Can workers reach the classifier within their deadline? Are completed decisions reaching the human-review queue exactly once? That order prevents a classifier investigation from hiding a database or queue problem. For recovery, select nonterminal reports by stored state and age, not by searching application logs. Logs are evidence, not a work queue.

Per-tenant visibility also changes admission control. Give each school a concurrency ceiling and a daily alert threshold, but preserve a small lane for urgent safety reports so a large batch import cannot starve interactive flags. Limits should control spend and blast radius, not silently discard reports. When a tenant crosses a threshold, queue the work, notify the owner, and expose the delay to reviewers. Don't convert a finance guardrail into missing moderation.

The comparison that matters is architectural:

The catch is latency and operational overhead. A durable queue, sweeper, ledger, and reconciliation job add moving parts. This design is not suitable when content never becomes visible, the volume is tiny, and a human already reviews every item before release; a synchronous call with a strict timeout may be enough there. Stick with a batch when the corpus is static and no safety decision gates publication. At the other extreme, regulated evidence retention may require a dedicated policy engine and immutable audit store rather than the compact ledger described here. Your mileage may vary because school policy and data residency constraints change the boundary.

Deployment gates and the decision rule

Ship policy changes as versioned deployments. Evaluate a fixed, consented fixture set for text, image, mixed input, adversarial formatting, and borderline classroom speech. Record false allows and false blocks separately; one aggregate accuracy number hides the error that determines whether a learner is exposed or unfairly silenced. Roll out by tenant cohort, watch review rates and schema failures, and retain the previous policy version for rollback.

The go/no-go rule is plain: do not automate blocking until the team can replay a report, explain which policy and content hashes produced the decision, prove duplicate delivery has one effective result, and reconcile usage to a tenant. Start with review as the automated outcome. Promote categories to automatic block only after evaluation and an appeal path support it.

This is where scheduler discipline earns its keep. Classification quality matters, but an accurate classifier attached to an unaccountable retry loop can still miss reports, duplicate work, and make tenant bills impossible to defend. The contract, queue state, ledger, and human handoff are one system.

References

• MDN, "Using server-sent events": https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events

Further reading

• LangChain, "ChatOpenAI integration": https://python.langchain.com/docs/integrations/chat/openai/