用 Scala 通过 Rovela API 发送邮件

本文介绍如何用 Scala 调用 Rovela API 发送事务邮件，涵盖标准库 HttpURLConnection、sttp 客户端、幂等键重试、parTraverse 并发批量发送，以及 http4s 服务端的 Webhook 验签。

前提条件

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

export ROVELA_API_KEY="re_..."

用 HttpURLConnection 发送

仅依赖 JDK：

import java.net.{HttpURLConnection, URL}
import scala.io.Source
import scala.util.{Try, Using}

object Rovela {
  private val apiUrl = "https://api.rovela.dev/emails"

  def sendEmail(to: String, subject: String, html: String): Try[String] = Try {
    val apiKey = sys.env("ROVELA_API_KEY")
    val conn = new URL(apiUrl).openConnection().asInstanceOf[HttpURLConnection]
    conn.setRequestMethod("POST")
    conn.setRequestProperty("Authorization", s"Bearer $apiKey")
    conn.setRequestProperty("Content-Type", "application/json")
    conn.setDoOutput(true)

    val body = s"""{"from":"Acme <noreply@yourdomain.com>","to":"$to","subject":"$subject","html":"$html"}"""
    Using.resource(conn.getOutputStream)(_.write(body.getBytes("UTF-8")))

    val code = conn.getResponseCode
    val stream = if (code >= 200 && code < 300) conn.getInputStream else conn.getErrorStream
    Using.resource(Source.fromInputStream(stream))(_.mkString)
  }
}

用 sttp 发送

sttp 是 Scala 常用的 HTTP 客户端，支持多种后端：

libraryDependencies ++= Seq(
  "com.softwaremill.sttp.client3" %% "core"          % "3.9.7",
  "com.softwaremill.sttp.client3" %% "cats"          % "3.9.7",
  "io.circe"                      %% "circe-generic" % "0.14.9",
  "org.typelevel"                 %% "cats-effect"   % "3.5.4"
)

import sttp.client3._
import sttp.client3.httpclient.cats.HttpClientCatsBackend
import cats.effect.IO
import io.circe.generic.semiauto._
import io.circe.syntax._
import io.circe._

case class SendEmailRequest(from: String, to: String, subject: String, html: String)

object RovelaSttp {
  implicit val encoder: Encoder[SendEmailRequest] = deriveEncoder

  def sendEmail(req: SendEmailRequest): IO[Response[String]] = {
    val apiKey = sys.env("ROVELA_API_KEY")

    // resource() 会真正建立一个基于 HttpClient 的后端
    HttpClientCatsBackend.resource[IO]().use { backend =>
      basicRequest
        .post(uri"https://api.rovela.dev/emails")
        .header("Authorization", s"Bearer $apiKey")
        .header("Content-Type", "application/json")
        .body(req.asJson.noSpaces)
        .send(backend)
    }
  }
}

重试与幂等键

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

import java.net.{HttpURLConnection, URL}
import java.security.SecureRandom
import scala.annotation.tailrec
import scala.io.Source
import scala.util.{Try, Using}

object RovelaRetry {
  private val apiUrl = "https://api.rovela.dev/emails"
  private val maxRetries = 4

  private def randomKey(): String = {
    val bytes = new Array[Byte](16)
    new SecureRandom().nextBytes(bytes)
    bytes.map(b => f"$b%02x").mkString
  }

  /** 2xx 返回 Right(body)，其余返回 Left(状态码)。 */
  private def postOnce(apiKey: String, key: String, payload: String): Either[Int, String] = {
    val conn = new URL(apiUrl).openConnection().asInstanceOf[HttpURLConnection]
    conn.setRequestMethod("POST")
    conn.setRequestProperty("Authorization", s"Bearer $apiKey")
    conn.setRequestProperty("Content-Type", "application/json")
    conn.setRequestProperty("Idempotency-Key", key)
    conn.setDoOutput(true)
    Using.resource(conn.getOutputStream)(_.write(payload.getBytes("UTF-8")))

    val code = conn.getResponseCode
    if (code >= 200 && code < 300)
      Right(Using.resource(Source.fromInputStream(conn.getInputStream))(_.mkString))
    else Left(code)
  }

  @tailrec
  private def loop(attempt: Int, apiKey: String, key: String, payload: String): Try[String] =
    postOnce(apiKey, key, payload) match {
      case Right(body) => Try(body)
      case Left(429) if attempt < maxRetries =>
        Thread.sleep((math.pow(2, attempt) * 1000).toInt)
        loop(attempt + 1, apiKey, key, payload)
      case Left(code) =>
        Try(throw new RuntimeException(s"HTTP $code"))
    }

  def sendWithRetry(payload: String): Try[String] = Try {
    val apiKey = sys.env("ROVELA_API_KEY")
    val key = randomKey() // 循环前生成，重试时复用
    loop(0, apiKey, key, payload).get
  }
}

Future 并发批量发送

parTraverse 并发执行发送，用 cats-effect 的 Semaphore 限制同时进行的请求数：

import cats.effect.{IO, Semaphore}
import cats.syntax.all._

case class Recipient(email: String, subject: String, html: String)

def sendOne(r: Recipient): String = {
  import java.net.{HttpURLConnection, URL}
  import scala.io.Source
  import scala.util.Using

  val conn = new URL("https://api.rovela.dev/emails")
    .openConnection().asInstanceOf[HttpURLConnection]
  conn.setRequestMethod("POST")
  conn.setRequestProperty("Authorization", s"Bearer ${sys.env("ROVELA_API_KEY")}")
  conn.setRequestProperty("Content-Type", "application/json")
  conn.setDoOutput(true)

  val body =
    s"""{"from":"Acme <noreply@yourdomain.com>","to":"${r.email}","subject":"${r.subject}","html":"${r.html}"}"""
  Using.resource(conn.getOutputStream)(_.write(body.getBytes("UTF-8")))

  val code = conn.getResponseCode
  val stream = if (code >= 200 && code < 300) conn.getInputStream else conn.getErrorStream
  Using.resource(Source.fromInputStream(stream))(_.mkString)
}

def sendBatch(recipients: List[Recipient], concurrency: Int = 5): IO[List[String]] =
  Semaphore[IO](concurrency).flatMap { sem =>
    recipients.parTraverse { r =>
      sem.permit.use(_ => IO.blocking(sendOne(r)))
    }
  }

不用 cats-effect 时，把列表按 concurrency 分块，每组用 Future.traverse 并发，前一组完成再启动下一组。

http4s Webhook 验签

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

libraryDependencies ++= Seq(
  "org.http4s" %% "http4s-dsl"          % "0.23.26",
  "org.http4s" %% "http4s-blaze-server" % "0.23.16"
)

import javax.crypto.Mac
import javax.crypto.spec.SecretKeySpec

import cats.effect.{IO, IOApp}
import org.http4s._
import org.http4s.blaze.server.BlazeServerBuilder
import org.http4s.dsl.io._
import org.http4s.implicits._
import org.http4s.server.Router
import org.typelevel.ci._

import scala.util.Try

object RovelaWebhook {
  private def constantTimeEqual(a: Array[Byte], b: Array[Byte]): Boolean = {
    if (a.length != b.length) return false
    var result = 0; var i = 0
    while (i < a.length) { result |= a(i) ^ b(i); i += 1 }
    result == 0
  }

  private def parseSignature(header: String): Option[(String, String)] = {
    val parts = header.split(",").flatMap { part =>
      part.split("=", 2) match {
        case Array(k, v) => Some(k -> v)
        case _           => None
      }
    }.toMap
    for {
      t  <- parts.get("t")
      v1 <- parts.get("v1")
    } yield (t, v1)
  }

  private def verify(secret: String, header: String, rawBody: String): Boolean =
    parseSignature(header).exists { case (ts, v1Hex) =>
      Try {
        val mac = Mac.getInstance("HmacSHA256")
        mac.init(new SecretKeySpec(secret.getBytes("UTF-8"), "HmacSHA256"))
        val expected = mac.doFinal(s"$ts.$rawBody".getBytes("UTF-8"))
        // 奇数长度的十六进制会在最后一段抛异常，被外层的 Try 捕获
        val provided = v1Hex.grouped(2).map(Integer.parseInt(_, 16).toByte).toArray
        constantTimeEqual(expected, provided)
      }.getOrElse(false)
    }

  def routes: HttpRoutes[IO] = HttpRoutes.of[IO] {
    case req @ POST -> Root / "webhooks" / "rovela" =>
      val secret  = sys.env("ROVELA_WEBHOOK_SECRET")
      val sigHead = req.headers.get(ci"x-resend-signature").map(_.head.value)

      req.as[String].flatMap { rawBody =>
        sigHead match {
          case None                                   => BadRequest("missing signature")
          case Some(h) if !verify(secret, h, rawBody) => BadRequest("invalid signature")
          case Some(_) =>
            // 解析 body 并处理事件
            Ok("ok")
        }
      }
  }
}

object Main extends IOApp.Simple {
  def run: IO[Unit] =
    BlazeServerBuilder[IO]
      .bindHttp(8080, "0.0.0.0")
      .withHttpApp(Router("/" -> RovelaWebhook.routes).orNotFound)
      .serve
      .compile
      .drain
}

constantTimeEqual 逐字节异或再 OR，全通过才返回 true，防止通过响应时间推断签名内容。