用 Elixir 通过 Rovela API 发送邮件

本文介绍如何用 Elixir 调用 Rovela API 发送事务邮件，涵盖 HTTPoison、Finch、幂等键重试、Task.async_stream 并发批量发送，以及 Plug 和 Phoenix 的 Webhook 验签。

前提条件

将 API 密钥设置为环境变量：

export ROVELA_API_KEY="re_..."

在 mix.exs 中添加依赖：

defp deps do
  [
    {:httpoison, "~> 2.0"},
    {:jason, "~> 1.4"}
  ]
end

用 HTTPoison 发送

defmodule Rovela do
  @api_url "https://api.rovela.dev/emails"

  def send_email(to, subject, html) do
    api_key = System.fetch_env!("ROVELA_API_KEY")

    headers = [
      {"Authorization", "Bearer #{api_key}"},
      {"Content-Type", "application/json"}
    ]

    body = Jason.encode!(%{
      from: "Acme <noreply@yourdomain.com>",
      to: to, subject: subject, html: html
    })

    case HTTPoison.post(@api_url, body, headers) do
      {:ok, %{status_code: s, body: b}} when s in 200..299 -> {:ok, Jason.decode!(b)}
      {:ok, %{status_code: s, body: b}}                    -> {:error, {s, b}}
      {:error, %{reason: r}}                               -> {:error, r}
    end
  end
end

用 Finch 发送

Finch 基于连接池，适合生产环境。先在 Application Supervisor 中启动：

# application.ex
children = [{Finch, name: RovelaFinch}]

defmodule Rovela.Finch do
  @api_url "https://api.rovela.dev/emails"

  def send_email(to, subject, html) do
    api_key = System.fetch_env!("ROVELA_API_KEY")

    headers = [
      {"authorization", "Bearer #{api_key}"},
      {"content-type", "application/json"}
    ]

    body = Jason.encode!(%{
      from: "Acme <noreply@yourdomain.com>",
      to: to, subject: subject, html: html
    })

    :post
    |> Finch.build(@api_url, headers, body)
    |> Finch.request(RovelaFinch)
    |> case do
      {:ok, %{status: s, body: b}} when s in 200..299 -> {:ok, Jason.decode!(b)}
      {:ok, %{status: s, body: b}}                    -> {:error, {s, b}}
      {:error, reason}                                 -> {:error, reason}
    end
  end
end

重试与幂等键

在循环（递归）之前生成一次幂等键，每次重试都复用同一个键，防止服务器接受请求后网络断开导致重复发送。只对 429 重试：

defmodule Rovela.Retry do
  @api_url "https://api.rovela.dev/emails"
  @max_retries 4

  def send_with_retry(payload, attempt \\ 0, idempotency_key \\ nil) do
    key = idempotency_key || (:crypto.strong_rand_bytes(16) |> Base.encode16(case: :lower))
    api_key = System.fetch_env!("ROVELA_API_KEY")

    headers = [
      {"Authorization", "Bearer #{api_key}"},
      {"Content-Type", "application/json"},
      {"Idempotency-Key", key}
    ]

    case HTTPoison.post(@api_url, Jason.encode!(payload), headers) do
      {:ok, %{status_code: 429, headers: h}} when attempt < @max_retries ->
        delay =
          case List.keyfind(resp_headers, "retry-after", 0) do
            {_, v} ->
              # Retry-After 也可能是 HTTP 日期，解析不了就退回指数退避
              case Integer.parse(v) do
                {secs, _} -> secs * 1000
                :error    -> trunc(:math.pow(2, attempt) * 1000)
              end

            nil ->
              trunc(:math.pow(2, attempt) * 1000)
          end
        Process.sleep(delay)
        send_with_retry(payload, attempt + 1, key)

      {:ok, %{status_code: s, body: b}} when s in 200..299 -> {:ok, Jason.decode!(b)}
      {:ok, %{status_code: s, body: b}}                    -> {:error, {s, b}}
      {:error, %{reason: r}}                               -> {:error, r}
    end
  end
end

Task.async_stream 并发批量发送

Task.async_stream/3 的 max_concurrency 参数直接限制并发数，不需要手动管理信号量：

defmodule Rovela.Batch do
  def send_batch(recipients, concurrency \\ 5) do
    recipients
    |> Task.async_stream(
      fn %{email: email, subject: subject, html: html} ->
        case Rovela.send_email(email, subject, html) do
          {:ok, result} -> {:ok, email, result}
          {:error, err} -> {:error, email, err}
        end
      end,
      max_concurrency: concurrency,
      timeout: 30_000,
      on_timeout: :kill_task
    )
    |> Enum.map(fn
      {:ok, result}   -> result
      {:exit, reason} -> {:error, :timeout, reason}
    end)
  end
end

结果保持原始顺序，单个任务失败不会中断其余任务。

Plug 验签

Rovela 用 HMAC-SHA256 对每个事件签名。签名在 x-resend-signature 请求头中，格式为 t=<时间戳>,v1=<十六进制签名>。对 <时间戳>.<原始 body> 用 Webhook 密钥计算 HMAC，再用 :crypto.hash_equals/2 做常量时间比对。

defmodule MyApp.RovelaWebhookPlug do
  import Plug.Conn

  def init(opts), do: opts

  def call(conn, _opts) do
    secret = System.fetch_env!("ROVELA_WEBHOOK_SECRET")
    {:ok, raw_body, conn} = Plug.Conn.read_body(conn)
    sig_header = get_req_header(conn, "x-resend-signature") |> List.first("")

    with {:ok, timestamp, v1_hex} <- parse_signature(sig_header),
         {:ok, v1_bytes}          <- Base.decode16(v1_hex, case: :lower),
         :ok                      <- verify(secret, timestamp, raw_body, v1_bytes) do
      payload = Jason.decode!(raw_body)
      conn |> handle_event(payload) |> send_resp(200, "ok")
    else
      {:error, reason} -> conn |> send_resp(400, inspect(reason)) |> halt()
    end
  end

  defp parse_signature(header) do
    parts =
      header
      |> String.split(",")
      |> Enum.flat_map(fn part ->
        case String.split(part, "=", parts: 2) do
          [k, v] -> [{k, v}]
          _      -> []
        end
      end)
      |> Map.new()

    case {parts["t"], parts["v1"]} do
      {t, v1} when not is_nil(t) and not is_nil(v1) -> {:ok, t, v1}
      _ -> {:error, :malformed_signature}
    end
  end

  defp verify(secret, timestamp, raw_body, v1_bytes) do
    expected = :crypto.mac(:hmac, :sha256, secret, "#{timestamp}.#{raw_body}")
    if :crypto.hash_equals(expected, v1_bytes), do: :ok, else: {:error, :invalid_signature}
  end

  defp handle_event(conn, _payload), do: conn
end

在路由中挂载：

plug "/webhooks/rovela", MyApp.RovelaWebhookPlug

Phoenix 验签

Phoenix 默认的 JSON 解析器会消费 body，需要先缓存原始字节，再在控制器里验签：

# lib/my_app/cache_body_reader.ex
defmodule MyApp.CacheBodyReader do
  def read_body(conn, opts) do
    {:ok, body, conn} = Plug.Conn.read_body(conn, opts)
    conn = Plug.Conn.put_private(conn, :raw_body, body)
    {:ok, body, conn}
  end
end

在 endpoint.ex 中把 Plug.Parsers 的 body_reader 换成上面的模块：

plug Plug.Parsers,
  parsers: [:urlencoded, :multipart, :json],
  pass: ["*/*"],
  body_reader: {MyApp.CacheBodyReader, :read_body, []},
  json_decoder: Jason

控制器：

defmodule MyAppWeb.WebhookController do
  use MyAppWeb, :controller

  def rovela(conn, _params) do
    secret     = System.fetch_env!("ROVELA_WEBHOOK_SECRET")
    raw_body   = conn.private[:raw_body] || ""
    sig_header = get_req_header(conn, "x-resend-signature") |> List.first("")

    with {:ok, ts, v1_hex} <- parse_sig(sig_header),
         {:ok, v1}         <- Base.decode16(v1_hex, case: :lower),
         expected          = :crypto.mac(:hmac, :sha256, secret, "#{ts}.#{raw_body}"),
         true              <- :crypto.hash_equals(expected, v1) do
      handle_event(conn.body_params["type"])
      send_resp(conn, 200, "ok")
    else
      _ -> send_resp(conn, 400, "invalid signature")
    end
  end

  defp parse_sig(header) do
    parts = for kv <- String.split(header, ","),
                [k, v] <- [String.split(kv, "=", parts: 2)],
                into: %{}, do: {k, v}
    case {parts["t"], parts["v1"]} do
      {t, v} when not is_nil(t) and not is_nil(v) -> {:ok, t, v}
      _ -> {:error, :malformed}
    end
  end

  defp handle_event(_type), do: :ok
end

路由：

scope "/webhooks" do
  post "/rovela", WebhookController, :rovela
end

:crypto.hash_equals/2 做常量时间比对，防止时序攻击。