Documentation

Backend pipeline with Phoenix

Receive AI webhooks in Phoenix controllers, process with Oban background jobs, and push results to LiveView clients in real time.

Overview

Phoenix's real-time capabilities: channels, LiveView, and PubSub: make it a natural fit for backend pipeline workflows. Receive webhooks in a controller, process with Oban, and push the final result to connected LiveView clients instantly.

What you'll build:

  • A Phoenix controller endpoint for receiving webhooks
  • HMAC signature verification plug
  • Oban worker for background processing
  • LiveView integration for real-time results

Quick start

Create project

Bash
mix phx.new my_ai_app
cd my_ai_app

Configuration

ELIXIR
1# config/runtime.exs
2config :my_ai_app,
3 modelriver_api_key: System.get_env("MODELRIVER_API_KEY"),
4 modelriver_webhook_secret: System.get_env("MODELRIVER_WEBHOOK_SECRET")

Dependencies

ELIXIR
1# mix.exs
2defp deps do
3 [
4 {:oban, "~> 2.17"},
5 {:req, "~> 0.5"},
6 # ... other deps
7 ]
8end

Signature verification plug

Plug.Parsers consumes the request body before your controller runs, so first keep a copy of the raw body with a custom body_reader:

ELIXIR
1# lib/my_ai_app_web/cache_body_reader.ex
2defmodule MyAiAppWeb.CacheBodyReader do
3 # Keep the exact raw bytes: Plug.Parsers consumes the body, but the signature is computed over it
4 def read_body(conn, opts) do
5 with {:ok, body, conn} <- Plug.Conn.read_body(conn, opts) do
6 {:ok, body, Plug.Conn.assign(conn, :raw_body, body)}
7 end
8 end
9end
10 
11# lib/my_ai_app_web/endpoint.ex
12plug Plug.Parsers,
13 parsers: [:urlencoded, :multipart, :json],
14 pass: ["*/*"],
15 body_reader: {MyAiAppWeb.CacheBodyReader, :read_body, []},
16 json_decoder: Phoenix.json_library()
ELIXIR
1# lib/my_ai_app_web/plugs/verify_modelriver_signature.ex
2defmodule MyAiAppWeb.Plugs.VerifyModelRiverSignature do
3 import Plug.Conn
4 require Logger
5 
6 def init(opts), do: opts
7 
8 def call(conn, _opts) do
9 signature = get_req_header(conn, "x-modelriver-signature") |> List.first("")
10 timestamp = get_req_header(conn, "x-modelriver-timestamp") |> List.first("")
11 secret = Application.get_env(:my_ai_app, :modelriver_webhook_secret)
12 raw_body = conn.assigns[:raw_body] || ""
13 
14 if valid_signature?(raw_body, timestamp, signature, secret) do
15 conn
16 else
17 conn
18 |> put_status(:unauthorized)
19 |> Phoenix.Controller.json(%{error: "Invalid signature"})
20 |> halt()
21 end
22 end
23 
24 # Signed string: "<timestamp>.<exact JSON text of the raw body's "data" value>" (never re-encode)
25 defp valid_signature?(raw_body, timestamp, signature, secret) do
26 with {start, len} <- :binary.match(raw_body, ~s("data":)),
27 [_ | _] = matches <- :binary.matches(raw_body, ~s(,"timestamp":)),
28 {finish, _} when finish >= start + len <- List.last(matches) do
29 data = binary_part(raw_body, start + len, finish - start - len)
30 
31 expected =
32 :crypto.mac(:hmac, :sha256, secret, "#{timestamp}.#{data}")
33 |> Base.encode16(case: :lower)
34 
35 Plug.Crypto.secure_compare(expected, signature)
36 else
37 _ -> false
38 end
39 end
40end

Webhook controller

ELIXIR
1# lib/my_ai_app_web/controllers/webhook_controller.ex
2defmodule MyAiAppWeb.WebhookController do
3 use MyAiAppWeb, :controller
4 require Logger
5 
6 plug MyAiAppWeb.Plugs.VerifyModelRiverSignature
7 
8 def modelriver(conn, params) do
9 payload = params["data"] || %{} # the event payload is under "data"
10 
11 case {payload["type"], payload["callback_url"]} do
12 {"task.ai_generated", callback_url} when is_binary(callback_url) ->
13 # Queue background job
14 %{
15 "event" => payload["event"],
16 "ai_response" => payload["ai_response"],
17 "callback_url" => callback_url,
18 "customer_data" => payload["customer_data"] || %{}
19 }
20 |> MyAiApp.Workers.ProcessAiWebhook.new()
21 |> Oban.insert()
22 
23 json(conn, %{received: true})
24 
25 _ ->
26 Logger.info("Standard webhook received: #{payload["status"]}")
27 json(conn, %{received: true})
28 end
29 end
30end

Router

ELIXIR
1# lib/my_ai_app_web/router.ex
2scope "/webhooks", MyAiAppWeb do
3 pipe_through :api
4 post "/modelriver", WebhookController, :modelriver
5end

Oban worker

ELIXIR
1# lib/my_ai_app/workers/process_ai_webhook.ex
2defmodule MyAiApp.Workers.ProcessAiWebhook do
3 use Oban.Worker, queue: :webhooks, max_attempts: 3
4 require Logger
5 
6 @impl Oban.Worker
7 def perform(%Oban.Job{args: args}) do
8 event = args["event"]
9 ai_response = args["ai_response"]
10 callback_url = args["callback_url"]
11 customer_data = args["customer_data"]
12 api_key = Application.get_env(:my_ai_app, :modelriver_api_key)
13 
14 enriched_data = ai_response["data"] || %{}
15 
16 # Your custom business logic
17 enriched_data =
18 case event do
19 "content_ready" ->
20 {:ok, content} =
21 MyAiApp.Content.create(%{
22 title: enriched_data["title"],
23 body: enriched_data["description"],
24 category: customer_data["category"] || "general"
25 })
26 
27 Map.merge(enriched_data, %{
28 "id" => content.id,
29 "slug" => content.slug,
30 "saved_at" => DateTime.utc_now() |> DateTime.to_iso8601()
31 })
32 
33 _ ->
34 enriched_data
35 end
36 
37 # Call back to ModelRiver
38 case Req.post(callback_url,
39 json: %{
40 data: enriched_data,
41 task_id: "phoenix_#{event}_#{System.system_time(:second)}",
42 metadata: %{
43 processed_by: "phoenix",
44 processed_at: DateTime.utc_now() |> DateTime.to_iso8601()
45 }
46 },
47 headers: [
48 {"authorization", "Bearer #{api_key}"},
49 {"content-type", "application/json"}
50 ],
51 receive_timeout: 10_000
52 ) do
53 {:ok, %{status: status}} when status in 200..299 ->
54 Logger.info("✅ Callback sent for event: #{event}")
55 :ok
56 
57 {:ok, %{status: status, body: body}} ->
58 Logger.error("❌ Callback failed (#{status}): #{inspect(body)}")
59 {:error, "Callback returned #{status}"}
60 
61 {:error, error} ->
62 Logger.error("❌ Callback error: #{inspect(error)}")
63 {:error, error}
64 end
65 end
66end

LiveView integration

Push the final result to connected LiveView clients using PubSub:

ELIXIR
1# In your Oban worker, after successful callback:
2Phoenix.PubSub.broadcast(
3 MyAiApp.PubSub,
4 "ai_results:#{customer_data["user_id"]}",
5 {:ai_result, enriched_data}
6)
7 
8# In your LiveView:
9defmodule MyAiAppWeb.ContentLive do
10 use MyAiAppWeb, :live_view
11 
12 def mount(_params, session, socket) do
13 if connected?(socket) do
14 user_id = session["user_id"]
15 Phoenix.PubSub.subscribe(MyAiApp.PubSub, "ai_results:#{user_id}")
16 end
17 
18 {:ok, assign(socket, result: nil, status: "idle")}
19 end
20 
21 def handle_info({:ai_result, data}, socket) do
22 {:noreply, assign(socket, result: data, status: "complete")}
23 end
24 
25 def render(assigns) do
26 ~H"""
27 <div>
28 <p :if={@status == "processing"}>Processing AI request...</p>
29 <pre :if={@result}><%= Jason.encode!(@result, pretty: true) %></pre>
30 </div>
31 """
32 end
33end

Best practices

  1. Use Oban: Background jobs with retries and telemetry.
  2. Use PubSub: Push results to LiveView clients without polling.
  3. Use Plug.Crypto.secure_compare: Constant-time signature verification.
  4. Use Req: Modern, composable HTTP client with built-in retry support.
  5. Cache the raw body: Plug.Parsers consumes the body, so use a body_reader that stores the raw bytes in conn.assigns for signature verification.

Next steps