用 Lua 通过 Rovela API 发送邮件

Lua 支撑着 OpenResty、Redis 脚本以及大量嵌入式场景，用它发事务邮件也完全不比那些重型运行时费劲。本文涵盖 LuaSocket + LuaSec、lua-http 和 OpenResty 三种发送方式，幂等键重试、copas 并发批量发送，以及 lua-resty-openssl 和 luaossl 的 Webhook 验签。

前提条件

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

export ROVELA_API_KEY="re_..."

用 LuaRocks 安装模块：

luarocks install luasocket luasec lua-cjson copas lua-http luaossl
# OpenResty 用户改从 opm 装 lua-resty-http 和 lua-resty-openssl：
opm get ledgetech/lua-resty-http
opm get fffonion/lua-resty-openssl

用 LuaSocket 发送

socket.http 不支持 TLS，拿 https:// 地址跑它要么静默失败、要么直接报错。要用 LuaSec 的 ssl.https：

local https = require "ssl.https"   -- LuaSec；socket.http 不支持 TLS
local ltn12 = require "ltn12"
local cjson = require "cjson.safe"

local payload = assert(cjson.encode({
  from    = "Acme <noreply@yourdomain.com>",
  to      = "user@example.com",
  subject = "Welcome aboard",
  html    = "<p>Thanks for signing up!</p>",
}))

local chunks = {}
local ok, code, headers = https.request {
  url    = "https://api.rovela.dev/emails",
  method = "POST",
  headers = {
    ["Authorization"]  = "Bearer " .. assert(os.getenv "ROVELA_API_KEY"),
    ["Content-Type"]   = "application/json",
    ["Content-Length"] = tostring(#payload),
  },
  source = ltn12.source.string(payload),
  sink   = ltn12.sink.table(chunks),
}

local response = table.concat(chunks)
if not ok or code < 200 or code > 299 then
  error(("%s: %s"):format(tostring(code), response))
end

print(assert(cjson.decode(response)).id)

用 lua-http 发送

lua-http 基于 cqueues，不用框架也能拿到非阻塞 I/O：

local http_request = require "http.request"
local cjson = require "cjson.safe"

local req = assert(http_request.new_from_uri("https://api.rovela.dev/emails"))
req.headers:upsert(":method", "POST")
req.headers:upsert("authorization", "Bearer " .. assert(os.getenv "ROVELA_API_KEY"))
req.headers:upsert("content-type", "application/json")
assert(req:set_body(assert(cjson.encode({
  from    = "Acme <noreply@yourdomain.com>",
  to      = "user@example.com",
  subject = "Welcome aboard",
  html    = "<p>Thanks for signing up!</p>",
}))))

local headers, stream = assert(req:go())
local body = assert(stream:get_body_as_string())
assert(headers:get ":status" == "200", body)

print(assert(cjson.decode(body)).id)

用 OpenResty 发送

在 OpenResty 里，lua-resty-http 会复用连接池，并遵循 nginx.conf 的 SSL 配置：

local httpc = require("resty.http").new()
local cjson = require "cjson.safe"

httpc:set_timeouts(2000, 2000, 5000)

local res, err = httpc:request_uri("https://api.rovela.dev/emails", {
  method = "POST",
  body   = cjson.encode({
    from    = "Acme <noreply@yourdomain.com>",
    to      = "user@example.com",
    subject = "Welcome aboard",
    html    = "<p>Thanks for signing up!</p>",
  }),
  headers = {
    ["Authorization"] = "Bearer " .. assert(os.getenv "ROVELA_API_KEY"),
    ["Content-Type"]  = "application/json",
  },
  ssl_verify = true,   -- 需要 nginx.conf 里配置 lua_ssl_trusted_certificate
})

httpc:close()

if not res then
  ngx.log(ngx.ERR, "rovela request failed: ", err)
  return ngx.exit(ngx.HTTP_BAD_GATEWAY)
end
if res.status < 200 or res.status > 299 then
  ngx.log(ngx.ERR, "rovela send failed: ", res.status, " ", res.body)
  return ngx.exit(ngx.HTTP_BAD_GATEWAY)
end

ngx.say(assert(cjson.decode(res.body)).id)

重试与幂等键

在循环之前生成一次幂等键，每次重试都复用同一个键，防止服务器已经接受请求后连接断开导致重复发送。只对 429 重试，其余状态码一律交给调用方：

local https  = require "ssl.https"
local ltn12  = require "ltn12"
local socket = require "socket"
local cjson  = require "cjson.safe"

local MAX_RETRIES = 4

local function idempotency_key()
  local f = assert(io.open("/dev/urandom", "rb"))
  local bytes = assert(f:read(16))
  f:close()
  return (bytes:gsub(".", function(c) return ("%02x"):format(c:byte()) end))
end

local function post(api_key, key, payload)
  local chunks = {}
  local ok, code, headers = https.request {
    url    = "https://api.rovela.dev/emails",
    method = "POST",
    headers = {
      ["Authorization"]   = "Bearer " .. api_key,
      ["Content-Type"]    = "application/json",
      ["Content-Length"]  = tostring(#payload),
      ["Idempotency-Key"] = key,
    },
    source = ltn12.source.string(payload),
    sink   = ltn12.sink.table(chunks),
  }
  return ok, code, headers, table.concat(chunks)
end

local api_key = assert(os.getenv "ROVELA_API_KEY")
local payload = assert(cjson.encode({
  from    = "Acme <noreply@yourdomain.com>",
  to      = "user@example.com",
  subject = "Welcome aboard",
  html    = "<p>Thanks for signing up!</p>",
}))

local key = idempotency_key()   -- 循环前生成一次，重试时复用
local body

for attempt = 0, MAX_RETRIES - 1 do
  local ok, code, headers, response = post(api_key, key, payload)

  if not ok then
    error("network error: " .. tostring(code))
  elseif code == 429 then
    socket.sleep(tonumber((headers or {})["retry-after"]) or 2 ^ attempt)
  elseif code >= 200 and code <= 299 then
    body = response
    break
  else
    error(("%d: %s"):format(code, response))
  end
end

if not body then
  error("rate limit exceeded after " .. MAX_RETRIES .. " attempts")
end

print(assert(cjson.decode(body)).id)

copas 并发批量发送

copas 把阻塞的 LuaSocket 调用变成协程，等网络时自动让出。它会把所有线程同时跑起来，所以要按批次切分列表，控制同时在途的请求数：

local copas = require "copas"
local https = require "ssl.https"
local ltn12 = require "ltn12"
local cjson = require "cjson.safe"

local CONCURRENCY = 5
local api_key = assert(os.getenv "ROVELA_API_KEY")

local function send_one(r)
  local payload = assert(cjson.encode({
    from    = "Acme <noreply@yourdomain.com>",
    to      = r.to,
    subject = r.subject,
    html    = r.html,
  }))
  local chunks = {}
  local ok, code = https.request {
    url    = "https://api.rovela.dev/emails",
    method = "POST",
    headers = {
      ["Authorization"]  = "Bearer " .. api_key,
      ["Content-Type"]   = "application/json",
      ["Content-Length"] = tostring(#payload),
    },
    source = ltn12.source.string(payload),
    sink   = ltn12.sink.table(chunks),
  }
  print(("%s: %s"):format(r.to, ok and tostring(code) or "error"))
end

local recipients = {
  { to = "ada@example.com",   subject = "Hello Ada",   html = "<p>Hi Ada</p>" },
  { to = "grace@example.com", subject = "Hello Grace", html = "<p>Hi Grace</p>" },
}

for i = 1, #recipients, CONCURRENCY do
  for j = i, math.min(i + CONCURRENCY - 1, #recipients) do
    local r = recipients[j]
    copas.addthread(function() send_one(r) end)
  end
  copas.loop()   -- 这一批跑完才返回
end

OpenResty Webhook 验签

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

local hmac  = require "resty.openssl.hmac"
local cjson = require "cjson.safe"

local function unhex(s)
  if #s % 2 ~= 0 or s:find("%X") then return nil end
  return (s:gsub("%x%x", function(pair)
    return string.char(tonumber(pair, 16))
  end))
end

local function constant_time_eq(a, b)
  if #a ~= #b then return false end
  local diff = 0
  for i = 1, #a do
    diff = diff + math.abs(a:byte(i) - b:byte(i))
  end
  return diff == 0
end

ngx.req.read_body()
local raw = ngx.req.get_body_data()
if not raw then
  -- body 已被缓冲到磁盘，调大 client_body_buffer_size
  return ngx.exit(ngx.HTTP_INTERNAL_SERVER_ERROR)
end

local header = ngx.req.get_headers()["x-resend-signature"] or ""
local ts, v1 = header:match("^t=([^,]+),v1=(%x+)$")
if not ts then
  return ngx.exit(ngx.HTTP_BAD_REQUEST)
end

local provided = unhex(v1)
if not provided then
  return ngx.exit(ngx.HTTP_BAD_REQUEST)
end

local mac = assert(hmac.new(ngx.var.rovela_webhook_secret, "sha256"))
local expected = assert(mac:final(ts .. "." .. raw))

if not constant_time_eq(expected, provided) then
  return ngx.exit(ngx.HTTP_UNAUTHORIZED)
end

local event = assert(cjson.decode(raw))
if event.type == "email.delivered" then
  -- 标记为已投递
elseif event.type == "email.bounced" then
  -- 标记该地址退信
end

ngx.say("ok")

luaossl Webhook 验签

不在 OpenResty 里时，luaossl 提供同样的原语。把校验写成与框架无关的函数，交给它原样的请求体：

local hmac = require "openssl.hmac"

local function unhex(s)
  if #s % 2 ~= 0 or s:find("%X") then return nil end
  return (s:gsub("%x%x", function(pair)
    return string.char(tonumber(pair, 16))
  end))
end

local function constant_time_eq(a, b)
  if #a ~= #b then return false end
  local diff = 0
  for i = 1, #a do
    diff = diff + math.abs(a:byte(i) - b:byte(i))
  end
  return diff == 0
end

-- 传入原始请求体，不要传解码后的表。
local function verify_rovela(secret, signature_header, raw_body)
  local ts, v1 = signature_header:match("^t=([^,]+),v1=(%x+)$")
  if not ts then
    return false, "missing signature fields"
  end

  local provided = unhex(v1)
  if not provided then
    return false, "malformed hex"
  end

  local expected = hmac.new(secret, "sha256"):final(ts .. "." .. raw_body)
  if not constant_time_eq(expected, provided) then
    return false, "invalid signature"
  end

  return true
end

用 lua-http 的服务端时，body 已经在缓冲区里：先调用 verify_rovela(secret, stream.request_headers:get("x-resend-signature") or "", assert(stream:get_body_as_string()))，再去解析它。

两处比较都会走完整个字符串才看结果，所以运行时间不会泄露前面有多少字节匹配。