Send email from DaloyJS with Cloudflare Email Service

{
  "$schema": "node_modules/wrangler/config-schema.json",
  "name": "acme-api",
  "main": "src/index.ts",
  "compatibility_date": "2026-10-01",
  "compatibility_flags": ["nodejs_compat"],
  "send_email": [
    // Mail to users: may only send From the one no-reply address.
    { "name": "EMAIL", "allowed_sender_addresses": ["no-reply@mail.acme.example.com"] },
    // Contact form: can only ever reach the support inbox.
    { "name": "CONTACT_EMAIL", "destination_address": "support@acme.example.com" },
  ],
  "queues": {
    "producers": [{ "binding": "EMAIL_QUEUE", "queue": "email-send" }],
    "consumers": [
      { "queue": "email-send", "max_retries": 5, "dead_letter_queue": "email-send-dlq" },
      { "queue": "email-events", "max_retries": 5 },
    ],
  },
  "vars": {
    "EMAIL_FROM": "no-reply@mail.acme.example.com",
    "EMAIL_FROM_NAME": "Acme",
    "CONTACT_INBOX": "support@acme.example.com",
    "SUPPORT_FORWARD_TO": "team@acme.example.com",
  },
}

import { env } from "cloudflare:workers";
import type { EmailSender } from "../plugins/email.ts";

export const sender: EmailSender = {
  async send(msg) {
    // The binding takes no AbortSignal; Worker CPU and wall-time limits bound the call.
    const { messageId } = await env.EMAIL.send({
      from: { email: env.EMAIL_FROM, name: env.EMAIL_FROM_NAME },
      to: msg.to,
      replyTo: msg.replyTo,
      subject: msg.subject,
      text: msg.text,
      html: msg.html,
      headers: msg.headers, // only allowlisted headers; List-Unsubscribe is one
    });
    return { id: messageId };
  },
};

import { App, secureHeaders } from "@daloyjs/core";
import { emailPlugin } from "./plugins/email.ts";
import { sender } from "./email/cloudflare.ts";
import { contactRoutes } from "./routes/contact.ts";
import { receiptRoutes } from "./routes/receipts.ts";

export const app = new App();
app.use(secureHeaders());
emailPlugin(app, sender);
contactRoutes(app);
receiptRoutes(app);

import { env } from "cloudflare:workers";
import { z } from "zod";
import { App, HttpError, rateLimit } from "@daloyjs/core";
import { escapeHtml } from "../plugins/email.ts";
import { headerSafe } from "../schemas/email.ts";

export function contactRoutes(app: App) {
  app.post(
    "/contact",
    {
      operationId: "submitContact",
      // Per isolate; back it with a shared store for a global limit (see the overview).
      hooks: rateLimit({ windowMs: 10 * 60_000, max: 5 }),
      request: {
        body: z
          .object({ name: headerSafe.max(80), email: z.email(), message: z.string().min(1).max(5000) })
          .strict(),
      },
      responses: {
        202: { description: "Accepted", body: z.object({ ok: z.literal(true) }) },
      },
    },
    async ({ body, state }) => {
      try {
        await env.CONTACT_EMAIL.send({
          to: env.CONTACT_INBOX, // the binding refuses any other recipient anyway
          from: { email: env.EMAIL_FROM, name: "Acme contact form" },
          replyTo: body.email, // the visitor, only as Reply-To
          subject: `Contact form: ${body.name}`,
          text: `From ${body.name} <${body.email}>\n\n${body.message}`,
          html: `<p>From ${escapeHtml(body.name)} &lt;${escapeHtml(body.email)}&gt;</p><p>${escapeHtml(body.message)}</p>`,
        });
      } catch (err) {
        state.log.error({ err }, "contact email failed"); // never echo provider errors
        throw new HttpError(502, { title: "Email could not be sent" });
      }
      return { status: 202, body: { ok: true as const } };
    },
  );
}

import { env } from "cloudflare:workers";
import { jobIdempotencyKey } from "@daloyjs/core";
import type { EmailMessage } from "../plugins/email.ts";

export type QueuedEmail = Omit<EmailMessage, "signal" | "idempotencyKey">;
export interface QueuedSend {
  key: string;
  msg: QueuedEmail;
}

/** Producer: call from a route after the server has looked up the recipient and built the body. */
export async function enqueueEmail(businessKey: string, msg: QueuedEmail): Promise<string> {
  const key = jobIdempotencyKey({ name: "email.send", key: businessKey });
  await env.EMAIL_QUEUE.send({ key, msg } satisfies QueuedSend); // body limit: 128 KB
  return key;
}

import { sender } from "./cloudflare.ts";
import type { QueuedSend } from "./queue.ts";

// Codes that mean "fix the request": retrying will never succeed.
const PERMANENT = new Set([
  "E_VALIDATION_ERROR",
  "E_FIELD_MISSING",
  "E_TOO_MANY_RECIPIENTS",
  "E_TOO_MANY_ATTACHMENTS",
  "E_SENDER_NOT_VERIFIED",
  "E_SENDER_DOMAIN_NOT_AVAILABLE",
  "E_RECIPIENT_NOT_ALLOWED",
  "E_CONTENT_TOO_LARGE",
  "E_HEADER_NOT_ALLOWED",
  "E_HEADER_USE_API_FIELD",
  "E_HEADER_VALUE_INVALID",
]);

export async function handleSendBatch(batch: MessageBatch<QueuedSend>): Promise<void> {
  for (const message of batch.messages) {
    const { key, msg } = message.body;
    try {
      if ((await alreadySent(key)) || (await isUndeliverable(msg.to))) {
        message.ack();
        continue;
      }
      const { id } = await sender.send(msg);
      await recordSent(key, id);
      message.ack();
    } catch (err) {
      const code = (err as { code?: unknown }).code;
      if (code === "E_RECIPIENT_SUPPRESSED") {
        await markUndeliverable(msg.to.toLowerCase(), "suppressed");
        message.ack();
      } else if (typeof code === "string" && PERMANENT.has(code)) {
        console.error(JSON.stringify({ event: "email.rejected", code, key })); // no recipient, no provider text
        message.ack();
      } else {
        // Rate limit, daily quota, internal error: back off; after max_retries it lands in email-send-dlq.
        message.retry({ delaySeconds: Math.min(60 * 2 ** message.attempts, 3600) });
      }
    }
  }
}

// Your data layer, for example a D1 table with a unique key column.
declare function alreadySent(key: string): Promise<boolean>;
declare function recordSent(key: string, messageId: string): Promise<void>;
declare function isUndeliverable(email: string): Promise<boolean>;
declare function markUndeliverable(email: string, reason: string): Promise<void>;

pnpm wrangler queues subscription create email-events \
  --source email.sending \
  --events message.bounced,message.complained \
  --zone-id <ZONE_ID> \
  --domain mail.acme.example.com

import { jobIdempotencyKey } from "@daloyjs/core";

export interface EmailSendingEvent {
  type: string;
  payload: { eventId: string; messageId: string; recipient: string; bounce?: { type: string } };
}

export async function handleEventBatch(batch: MessageBatch<EmailSendingEvent>): Promise<void> {
  for (const message of batch.messages) {
    const { type, payload } = message.body;
    const hard =
      (type === "cf.email.sending.message.bounced" && payload.bounce?.type === "hard") ||
      type === "cf.email.sending.message.complained";
    if (hard) {
      // A redelivered event has the same eventId, so the insert is a no-op the second time.
      const key = jobIdempotencyKey({ name: "email.suppress", key: `cf-${payload.eventId}` });
      await recordSuppression(key, payload.recipient.toLowerCase(), type);
    }
    message.ack();
  }
}

// Insert into your suppression table, ignoring a duplicate key.
declare function recordSuppression(key: string, email: string, reason: string): Promise<void>;

import PostalMime from "postal-mime";

export async function handleInbound(
  message: ForwardableEmailMessage,
  env: Env,
  ctx: ExecutionContext,
): Promise<void> {
  if (message.rawSize > 5 * 1024 * 1024) {
    message.setReject("Message too large"); // permanent SMTP rejection
    return;
  }
  const parsed = await PostalMime.parse(message.raw); // raw is a one-shot stream
  // Sender addresses and every header are untrusted input: never grant access based on them,
  // and escape the text before you render it anywhere.
  ctx.waitUntil(
    createTicket({ from: message.from, subject: parsed.subject ?? "(no subject)", text: parsed.text ?? "" }),
  );
  await message.forward(env.SUPPORT_FORWARD_TO); // must be a verified destination address
}

declare function createTicket(t: { from: string; subject: string; text: string }): Promise<void>;

import { toFetchHandler } from "@daloyjs/core/cloudflare";
import { app } from "./server.ts";
import { handleSendBatch } from "./email/consumer.ts";
import { handleEventBatch, type EmailSendingEvent } from "./email/events.ts";
import { handleInbound } from "./email/inbound.ts";
import type { QueuedSend } from "./email/queue.ts";

export default {
  ...toFetchHandler<Env>(app),
  async queue(batch) {
    if (batch.queue === "email-send") await handleSendBatch(batch as MessageBatch<QueuedSend>);
    else if (batch.queue === "email-events") await handleEventBatch(batch as MessageBatch<EmailSendingEvent>);
  },
  async email(message, env, ctx) {
    await handleInbound(message, env, ctx);
  },
} satisfies ExportedHandler<Env>;

const pdf = await env.INVOICES.get(`invoices/${invoice.id}.pdf`); // an R2 binding
if (pdf) {
  await env.EMAIL.send({
    from: { email: env.EMAIL_FROM, name: env.EMAIL_FROM_NAME },
    to: invoice.customerEmail, // looked up server-side
    subject: `Invoice ${invoice.number}`,
    text: "Your invoice is attached.",
    attachments: [
      { content: await pdf.arrayBuffer(), filename: `invoice-${invoice.number}.pdf`, type: "application/pdf", disposition: "attachment" },
    ],
  });
}

import type { EmailSender } from "../plugins/email.ts";
import { providerFetch } from "./fetch.ts"; // allowHosts: ["api.cloudflare.com"]
import { config } from "../config.ts";

export const sender: EmailSender = {
  async send(msg) {
    const res = await providerFetch(
      `https://api.cloudflare.com/client/v4/accounts/${config.CF_ACCOUNT_ID}/email/sending/send`,
      {
        method: "POST",
        headers: { authorization: `Bearer ${config.CF_EMAIL_TOKEN}`, "content-type": "application/json" },
        signal: msg.signal,
        body: JSON.stringify({
          from: { address: config.EMAIL_FROM_ADDRESS, name: "Acme" },
          to: msg.to,
          reply_to: msg.replyTo,
          subject: msg.subject,
          text: msg.text,
          html: msg.html,
          headers: msg.headers,
        }),
      },
    );
    const out = (await res.json()) as { success: boolean; errors?: { code: number }[]; result?: { permanent_bounces: string[] } };
    if (!res.ok || !out.success) throw new Error(`Cloudflare email error ${out.errors?.[0]?.code ?? res.status}`);
    for (const email of out.result?.permanent_bounces ?? []) await markUndeliverable(email.toLowerCase(), "hard_bounce");
    return { id: "" }; // the REST API returns per-recipient status, not a message id
  },
};

declare function markUndeliverable(email: string, reason: string): Promise<void>;