Phoenix：后端管道 – ModelRiver Docs

概述

Phoenix 的实时处理能力（如 Channels、LiveView 和 PubSub）使其非常适合后端管道的 AI 工作流。您可以在控制器中接收 Webhooks，使用 Oban 进行处理，并立即将最终结果推送到已连接的 LiveView 客户端。

您将构建的内容：

一个用于接收 Webhooks 的 Phoenix 控制器端点
HMAC 签名验证 Plug
用于后台处理的 Oban Worker
用于实时展示结果的 LiveView 集成

快速开始

创建项目

Bash

mix phx.new my_ai_app
cd my_ai_app

配置

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")

依赖项

ELIXIR

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

签名验证 Plug

Plug.Parsers 会在控制器运行之前读取并消耗请求体，因此需要先通过自定义 body_reader 保留一份原始请求体：

ELIXIR

1# lib/my_ai_app_web/cache_body_reader.ex
2defmodule MyAiAppWeb.CacheBodyReader do
3  # 保留原始字节：Plug.Parsers 会消耗请求体，而签名是基于原始请求体计算的
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  # 签名字符串："<timestamp>.<原始请求体中 "data" 值的 JSON 文本>"（切勿重新编码）
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 控制器

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"] || %{}  # 事件载荷位于 "data" 之下
10 
11    case {payload["type"], payload["callback_url"]} do
12      {"task.ai_generated", callback_url} when is_binary(callback_url) ->
13        # 将后台任务加入队列
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("收到标准 Webhook: #{payload["status"]}")
27        json(conn, %{received: true})
28    end
29  end
30end

路由

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    # 您的自定义业务逻辑
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    # 回调 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("✅ 已针对事件发送回调: #{event}")
55        :ok
56 
57      {:ok, %{status: status, body: body}} ->
58        Logger.error("❌ 回调失败 (#{status}): #{inspect(body)}")
59        {:error, "回调返回了 #{status}"}
60 
61      {:error, error} ->
62        Logger.error("❌ 回调错误: #{inspect(error)}")
63        {:error, error}
64    end
65  end
66end

LiveView 集成

使用 PubSub 将最终结果推送到已连接的 LiveView 客户端：

ELIXIR

1# 在您的 Oban worker 中，回调成功后：
2Phoenix.PubSub.broadcast(
3  MyAiApp.PubSub,
4  "ai_results:#{customer_data["user_id"]}",
5  {:ai_result, enriched_data}
6)
7 
8# 在您的 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"}>正在处理 AI 请求...</p>
29      <pre :if={@result}><%= Jason.encode!(@result, pretty: true) %></pre>
30    </div>
31    """
32  end
33end

最佳实践

使用 Oban：支持重试和遥测监控的后台任务。
使用 PubSub：无需轮询即可将结果推送给 LiveView 客户端。
使用 Plug.Crypto.secure_compare：恒定时间签名验证。
使用 Req：现代化的可组合 HTTP 客户端，内置重试支持。
缓存原始 Body：Plug.Parsers 会消耗请求体，因此请使用 body_reader 将原始字节存入 conn.assigns，以供签名验证使用。

下一步

Spring Boot 后端管道指南：Java 备选方案
Phoenix 集成：标准 ModelRiver + Phoenix 的用法
Webhooks 参考：重试策略和投递监控

Phoenix 的后端管道

概述

快速开始

创建项目

配置

依赖项

签名验证 Plug

Webhook 控制器

路由

Oban Worker

LiveView 集成

最佳实践

下一步