Content Moderation in Elixir: Screen User Content with the OpenAI Moderation API

The moment your Phoenix app lets people type into a box that other people will read, you have a moderation problem. Comments, chat messages, profile bios, forum posts, prompts sent to your AI feature: all of it is user-generated content, and some fraction of it will be abuse, harassment or spam the day you launch.

Most teams handle this badly in one of two ways. They write a word blocklist (which misses everything clever and blocks the town of Scunthorpe), or they ship nothing and plan to “deal with it when it happens”. This guide shows a third option that costs almost nothing: content moderation in Elixir with the OpenAI Moderation API, wired into Ecto changesets, Oban jobs and LiveView, with a review queue for the cases a model should not decide alone.

This is part of our series on building AI apps with Elixir and Phoenix. If you already run an LLM feature, read this alongside Prompt Injection in Elixir: injection is about protecting your system from the user, moderation is about protecting your users from each other (and from your model).

What the Moderation API gives you

OpenAI’s moderation endpoint takes text (and, with the omni model, images) and returns a classification across a fixed set of categories: harassment, hate, self-harm, sexual content, violence, illicit activity, and finer-grained subcategories such as harassment/threatening or sexual/minors.

Three properties make it a good default for a Phoenix app:

• It is free to use for checking content. You still need an API key, and you are still subject to rate limits, but moderation calls are not billed per token the way completions are.
• It returns scores, not just a verdict. Each category gets a float between 0 and 1. OpenAI also returns a flagged boolean, but the scores let you set your own thresholds per category and per surface.
• It is fast. A short comment usually comes back in a few hundred milliseconds, which is fine for an async check and tolerable for a synchronous one.

It also has real limits, which we will come back to: it is tuned for English first, it knows nothing about your community’s rules, and it cannot judge context (“I’m going to kill this deploy” is not a threat). Treat it as a first filter, not a judge.

Step 1: A small moderation client with Req

Start with a module that sends content and returns something your app can reason about. We use Req, which most Phoenix AI code already depends on.

defmodule MyApp.Moderation do
  @moduledoc "Screens user content with the OpenAI Moderation API."

  @endpoint "https://api.openai.com/v1/moderations"
  @model "omni-moderation-latest"

  @type result :: %{
          flagged: boolean(),
          categories: %{String.t() => boolean()},
          scores: %{String.t() => float()}
        }

  @spec check(String.t()) :: {:ok, result()} | {:error, term()}
  def check(text) when is_binary(text) do
    [url: @endpoint, json: %{model: @model, input: text}]
    |> Keyword.merge(Application.get_env(:my_app, :moderation_req_options, []))
    |> Req.new()
    |> Req.Request.put_header("authorization", "Bearer " <> api_key())
    |> Req.post(retry: :transient, receive_timeout: 10_000)
    |> case do
      {:ok, %Req.Response{status: 200, body: %{"results" => [r | _]}}} ->
        {:ok,
         %{
           flagged: r["flagged"],
           categories: r["categories"],
           scores: r["category_scores"]
         }}

      {:ok, %Req.Response{status: status, body: body}} ->
        {:error, {:http, status, body}}

      {:error, exception} ->
        {:error, exception}
    end
  end

  defp api_key, do: Application.fetch_env!(:my_app, :openai_api_key)
end

Two details are worth a sentence each. retry: :transient lets Req retry on 408, 429 and 5xx responses with backoff, which covers most blips (our post on OpenAI rate limits in Elixir goes deeper). And the :moderation_req_options config hook is there so tests can plug in Req.Test without touching the module, which we use in step 6.

Step 2: Your policy, not OpenAI’s

The flagged boolean reflects OpenAI’s default thresholds. It is a reasonable start, but it is not your policy. A gaming community tolerates trash talk that a children’s education app must not. A medical support forum needs to discuss self-harm without auto-removing every post about it.

Put your policy in one module, as data, so it is easy to read and easy to change:

defmodule MyApp.Moderation.Policy do
  @moduledoc "Turns raw moderation scores into a decision for one surface."

  # score >= block: reject outright
  # score >= review: accept, but hide until a human looks
  @policies %{
    comments: %{
      "harassment" => {0.5, 0.8},
      "harassment/threatening" => {0.2, 0.5},
      "hate" => {0.3, 0.6},
      "sexual" => {0.5, 0.8},
      "sexual/minors" => {0.01, 0.1},
      "violence" => {0.6, 0.9},
      "self-harm" => {0.3, 1.1}
    },
    profile_bio: %{
      "sexual" => {0.3, 0.6},
      "hate" => {0.2, 0.5},
      "sexual/minors" => {0.01, 0.1}
    }
  }

  @type decision :: :allow | {:review, [String.t()]} | {:block, [String.t()]}

  @spec decide(atom(), MyApp.Moderation.result()) :: decision()
  def decide(surface, %{scores: scores}) do
    rules = Map.fetch!(@policies, surface)

    {block, review} =
      Enum.reduce(rules, {[], []}, fn {category, {review_at, block_at}}, {b, r} ->
        score = Map.get(scores, category, 0.0)

        cond do
          score >= block_at -> {[category | b], r}
          score >= review_at -> {b, [category | r]}
          true -> {b, r}
        end
      end)

    cond do
      block != [] -> {:block, block}
      review != [] -> {:review, review}
      true -> :allow
    end
  end
end

Look at the self-harm rule for comments: the block threshold is 1.1, which a score can never reach. That is deliberate. Someone posting about self-harm in a support community should never be silently rejected; the right response is a human (and probably a resources link), so that category can only ever send a post to review.

Thresholds like these are a starting point. Log every decision with its scores for the first few weeks, look at what landed in review, and tune. The numbers above are illustrative, not recommendations for your community.

Step 3: Synchronous checks in a changeset

For short content where the user is waiting anyway (a profile bio, a display name, a team name), a synchronous check gives the best experience: the user sees the problem immediately and can fix it.

Ecto changesets are pure functions by convention, so do not call the network from inside changeset/2. Run the check in the context function, after the cheap validations pass, and add the error there:

defmodule MyApp.Accounts do
  alias MyApp.Accounts.Profile
  alias MyApp.{Moderation, Repo}
  alias MyApp.Moderation.Policy

  def update_profile(%Profile{} = profile, attrs) do
    changeset = Profile.changeset(profile, attrs)

    with true <- changeset.valid?,
         bio when is_binary(bio) <- Ecto.Changeset.get_change(changeset, :bio),
         {:ok, result} <- Moderation.check(bio) do
      case Policy.decide(:profile_bio, result) do
        :allow ->
          Repo.update(changeset)

        {_verdict, _categories} ->
          changeset
          |> Ecto.Changeset.add_error(:bio, "goes against our community guidelines")
          |> Ecto.Changeset.apply_action(:update)
      end
    else
      false -> Ecto.Changeset.apply_action(changeset, :update)
      nil -> Repo.update(changeset)
      {:error, _reason} -> moderation_unavailable(changeset)
    end
  end

  # Fail closed for short, public fields: ask the user to retry.
  defp moderation_unavailable(changeset) do
    changeset
    |> Ecto.Changeset.add_error(:bio, "could not be checked right now, please try again")
    |> Ecto.Changeset.apply_action(:update)
  end
end

Notice the error message. Do not tell the user which category tripped or what the score was. Specific feedback is a tuning manual for the people trying to get around you. “Goes against our community guidelines” plus a link to those guidelines is enough for an honest user.

Fail open or fail closed?

Every moderation integration has to answer one question: what happens when the API is down? There is no universal answer, but there is a useful rule:

• Fail closed (reject, ask to retry) for low-volume, high-visibility fields like profile bios and team names, where a short outage is a minor annoyance.
• Fail open into review for high-volume content like comments and chat, where blocking every message during an outage would take the product down with it. Accept the content, keep it hidden or limited, and queue it for a re-check.

Decide this per surface and write it down, because whoever is on call at 2am will need to know.

Step 4: Asynchronous moderation with Oban

For comments and posts, a synchronous check adds latency to every submission and turns an OpenAI outage into your outage. The better pattern is to save the content in a pending state, show it to its author immediately, and moderate in the background.

Add a status to the schema:

schema "comments" do
  field :body, :string
  field :moderation_status, Ecto.Enum,
    values: [:pending, :approved, :in_review, :rejected],
    default: :pending
  field :moderation_categories, {:array, :string}, default: []
  belongs_to :author, MyApp.Accounts.User
  belongs_to :post, MyApp.Blog.Post
  timestamps()
end

Then insert the comment and its job in one transaction, so you can never have a comment without a moderation job:

def create_comment(author, post, attrs) do
  Ecto.Multi.new()
  |> Ecto.Multi.insert(:comment, Comment.changeset(%Comment{author_id: author.id, post_id: post.id}, attrs))
  |> Oban.insert(:moderate, fn %{comment: c} ->
    MyApp.Workers.ModerateComment.new(%{comment_id: c.id})
  end)
  |> Repo.transaction()
end

The worker calls the API, applies the policy and broadcasts the result:

defmodule MyApp.Workers.ModerateComment do
  use Oban.Worker, queue: :moderation, max_attempts: 10

  alias MyApp.{Moderation, Repo}
  alias MyApp.Blog.Comment
  alias MyApp.Moderation.Policy

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"comment_id" => id}}) do
    with %Comment{moderation_status: :pending} = comment <- Repo.get(Comment, id),
         {:ok, result} <- Moderation.check(comment.body) do
      {status, categories} =
        case Policy.decide(:comments, result) do
          :allow -> {:approved, []}
          {:review, cats} -> {:in_review, cats}
          {:block, cats} -> {:rejected, cats}
        end

      comment
      |> Ecto.Changeset.change(moderation_status: status, moderation_categories: categories)
      |> Repo.update!()
      |> broadcast()

      :ok
    else
      nil -> :ok
      %Comment{} -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  defp broadcast(comment) do
    Phoenix.PubSub.broadcast(MyApp.PubSub, "post:#{comment.post_id}", {:comment_moderated, comment})
  end
end

Returning {:error, reason} on an API failure lets Oban retry with its default exponential backoff, so a ten-minute OpenAI incident resolves itself without anyone touching it. Ten attempts covers a long outage; after that the job lands in discarded, and an alert on discarded moderation jobs is cheap insurance.

The guard on moderation_status: :pending makes the job idempotent. If it runs twice, or a moderator already acted on the comment, the second run does nothing.

Who sees a pending comment?

This is a product decision, but the common answer is: the author sees their own comment instantly (so the app feels fast), and everyone else sees it once it is approved. In a query that is one clause:

def visible_comments(post, viewer) do
  from c in Comment,
    where: c.post_id == ^post.id,
    where: c.moderation_status == :approved or c.author_id == ^viewer.id,
    order_by: [asc: c.inserted_at]
end

For most comments the gap between pending and approved is under a second, so other readers rarely notice it.

Step 5: Moderating a LiveView chat, including the model’s output

A real-time chat changes the tradeoff, because a message nobody else can see for a second feels broken. A workable pattern is to moderate in a Task started from the LiveView and only broadcast to the room once the result comes back:

def handle_event("send", %{"body" => body}, socket) do
  id = System.unique_integer([:positive])

  Task.Supervisor.async_nolink(MyApp.TaskSupervisor, fn ->
    {id, MyApp.Moderation.check(body)}
  end)

  {:noreply,
   socket
   |> stream_insert(:messages, %{id: id, body: body, status: :sending})
   |> update(:pending, &Map.put(&1, id, body))}
end

def handle_info({task_ref, {id, result}}, socket) when is_reference(task_ref) do
  Process.demonitor(task_ref, [:flush])
  {body, pending} = Map.pop(socket.assigns.pending, id)
  %{room: room, current_user: user} = socket.assigns

  status =
    case result do
      {:ok, r} ->
        case MyApp.Moderation.Policy.decide(:comments, r) do
          :allow -> MyApp.Chat.publish(room, user, body) && :sent
          _ -> :held
        end

      {:error, _} ->
        MyApp.Chat.hold_for_review(room, user, body) && :held
    end

  {:noreply,
   socket
   |> assign(:pending, pending)
   |> stream_insert(:messages, %{id: id, body: body, status: status})}
end

The sender sees their message immediately as “sending”, the room sees it only once it passes, and an API failure holds it for review rather than leaking it. A held message stays visible to its author with a quiet “held for review” label, which is kinder (and less informative to a bad actor) than making it vanish.

If your chat includes an AI assistant, moderate the model’s output too. Models are trained to refuse harmful requests, but a determined user plus a long conversation can still coax out text you would not want on your platform, and “the AI wrote it” is not a defence anyone will accept. For streamed responses, buffer the completed reply and check it before you persist it, or check every few hundred characters while streaming if you would rather cut a reply short than retract it. Our guide on streaming OpenAI responses in LiveView shows where that buffer lives.

Step 6: Testing without calling OpenAI

You do not want your test suite hitting a real API. Req.Test lets you stub the endpoint per test. In config/test.exs:

config :my_app, :moderation_req_options, plug: {Req.Test, MyApp.Moderation}
config :my_app, :openai_api_key, "test-key"

Then in a test:

test "a threatening comment is rejected" do
  Req.Test.stub(MyApp.Moderation, fn conn ->
    Req.Test.json(conn, %{
      "results" => [
        %{
          "flagged" => true,
          "categories" => %{"harassment/threatening" => true},
          "category_scores" => %{"harassment/threatening" => 0.93, "harassment" => 0.88}
        }
      ]
    })
  end)

  comment = insert(:comment, body: "a nasty message", moderation_status: :pending)

  assert :ok = perform_job(MyApp.Workers.ModerateComment, %{comment_id: comment.id})
  assert %{moderation_status: :rejected} = Repo.reload!(comment)
end

Stub three cases for every surface: a clean result, a review-band result and a block-band result. Add a fourth that returns a 500 to prove your fail-open or fail-closed choice actually behaves the way you wrote it down. Policy.decide/2 is a pure function, so table-test it directly with plain maps; that is where most of the logic lives.

The review queue: where the model stops

Everything that lands in :in_review needs a human. The minimum viable queue is a LiveView that lists in_review items oldest first, shows the content and the flagged categories, and offers two buttons: approve and reject. Record who decided and when, because appeals and audits will ask.

Two cheap additions pay for themselves quickly:

• User reports. Add a “report” link that moves an approved item back to :in_review. The model will miss things your community will not.
• Repeat offenders. Count rejections per author. Three rejections in a day is a better signal for a temporary posting limit than any single score.

Honest tradeoffs

The Moderation API is a good default, not a complete solution. Be clear-eyed about where it falls short:

• Language coverage is uneven. It handles many languages, but quality is best in English. If your community is mostly non-English, sample results by hand before trusting the thresholds.
• It has no context. It scores a message in isolation. Sarcasm, quotes, reclaimed slurs and in-group jokes all confuse it. Your review queue exists for exactly this.
• It cannot enforce your rules. Spam, self-promotion, off-topic posts and doxxing do not map cleanly onto its categories. For those you still want rate limits, link limits and, for LLM-heavy apps, a cheap completion call with your own written guidelines.
• It is a third party. Every checked message goes to OpenAI. That is usually fine for public posts, but say so in your privacy policy, and think twice before sending private messages.
• Local alternatives exist. If data residency rules out a hosted API, a classifier run locally with Bumblebee keeps content on your servers, at the cost of hosting and tuning it yourself.

Skip the plumbing

Moderation is one piece of the work an AI feature needs before it is safe to ship: API clients, streaming, retries, background jobs, cost tracking and an admin area to review what users and models produced. phx_ai is a Phoenix starter with that groundwork in place, with OpenAI wired through Req, LiveView streaming and Oban, so you can add a moderation step like this one to a working app instead of a blank project.

Summary

Content moderation in Elixir comes down to a few well-placed pieces:

1. A small Req client for the OpenAI Moderation API that returns scores, not just a flag.
2. A policy module that turns those scores into allow, review or block, per surface, with your thresholds.
3. Synchronous checks for short, high-visibility fields, done in the context function rather than the changeset, failing closed.
4. Asynchronous checks with Oban for comments and posts, in the same transaction as the insert, failing open into review.
5. Moderation of model output, not just user input, for any AI feature.
6. Req.Test stubs for clean, review, block and outage cases.
7. A human review queue and user reports for everything a classifier cannot judge.

None of it is much code, and all of it is cheaper to build before launch than after the first incident.

If you are planning more than one Phoenix product, the Builder Pass gives you every PhxTemplates starter, including the AI templates, for a single lifetime price.