Send email from DaloyJS with Azure Communication Services

ACS_ENDPOINT=https://acme-comms.communication.azure.com
ACS_SENDER=no-reply@mail.acme.example.com
EMAIL_QUEUE_URL=https://acmestorage.queue.core.windows.net
CONTACT_INBOX=support@acme.example.com
# Identity-based connection for the queue triggers (no storage key):
EmailQueue__queueServiceUri=https://acmestorage.queue.core.windows.net
# Only for a user-assigned managed identity:
AZURE_CLIENT_ID=00000000-0000-0000-0000-000000000000

import { EmailClient } from "@azure/communication-email";
import { DefaultAzureCredential } from "@azure/identity";
import type { EmailSender } from "../plugins/email.ts";
import { config } from "../config.ts";

// Managed identity in Azure, developer credentials locally. Pin ManagedIdentityCredential
// in production if you want no fallback chain.
export const acs = new EmailClient(config.ACS_ENDPOINT, new DefaultAzureCredential());

const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;

export const sender: EmailSender = {
  async send(msg) {
    // ACS identifies the operation by a caller-supplied UUID. The queue message carries one,
    // so a retried message reuses it and the id in your logs stays stable.
    const operationId =
      msg.idempotencyKey !== undefined && UUID.test(msg.idempotencyKey) ? msg.idempotencyKey : crypto.randomUUID();
    const content =
      msg.html !== undefined
        ? { subject: msg.subject, html: msg.html, plainText: msg.text }
        : { subject: msg.subject, plainText: msg.text ?? "" };
    await acs.beginSend(
      {
        senderAddress: config.ACS_SENDER,
        content,
        recipients: { to: [{ address: msg.to }] },
        replyTo: msg.replyTo !== undefined ? [{ address: msg.replyTo }] : undefined,
        headers: msg.headers,
      },
      { operationId, abortSignal: msg.signal },
    );
    // beginSend() resolved: ACS accepted the request. Do not pollUntilDone() here;
    // the delivery outcome arrives as an Event Grid delivery report.
    return { id: operationId };
  },
};

import { App, secureHeaders } from "@daloyjs/core";
import { emailPlugin } from "./plugins/email.ts";
import { sender } from "./email/acs.ts";
import { verifyEmailRoutes } from "./routes/verify-email.ts";

export const app = new App({ behindProxy: { hops: 1 } }); // see the Azure adapter's client-IP notes
app.use(secureHeaders());
emailPlugin(app, sender);
verifyEmailRoutes(app);

import { app as azure } from "@azure/functions";
import { toAzureHandler } from "@daloyjs/core/azure";
import { app } from "../app.ts";

azure.http("api", {
  route: "{*path}",
  methods: ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"],
  authLevel: "anonymous", // DaloyJS does the auth
  handler: toAzureHandler(app),
});

import { QueueClient } from "@azure/storage-queue";
import { DefaultAzureCredential } from "@azure/identity";
import { jobIdempotencyKey } from "@daloyjs/core";
import type { EmailMessage } from "../plugins/email.ts";
import { config } from "../config.ts";

export type QueuedEmail = Omit<EmailMessage, "signal" | "idempotencyKey">;
export interface QueuedSend {
  key: string; // your dedupe key, from the business event
  operationId: string; // the ACS operation id, stable across retries of this message
  msg: QueuedEmail;
}

// The identity needs a data-plane send role on the queue (for example Storage Queue Data Message Sender).
const sendQueue = new QueueClient(`${config.EMAIL_QUEUE_URL}/email-send`, new DefaultAzureCredential());

export async function enqueueEmail(businessKey: string, msg: QueuedEmail): Promise<string> {
  const job: QueuedSend = {
    key: jobIdempotencyKey({ name: "email.send", key: businessKey }),
    operationId: crypto.randomUUID(),
    msg,
  };
  await sendQueue.sendMessage(Buffer.from(JSON.stringify(job)).toString("base64"));
  return job.key;
}

import { z } from "zod";
import { App, UnauthorizedError } from "@daloyjs/core";
import { escapeHtml } from "../plugins/email.ts";
import { sendGuards } from "../email/guards.ts";
import { enqueueEmail } from "../email/queue.ts";

export function verifyEmailRoutes(app: App) {
  app.post(
    "/me/verify-email",
    {
      operationId: "resendVerification",
      hooks: sendGuards, // auth + per-IP + per-user limits, see the overview
      responses: { 202: { description: "Queued", body: z.object({ ok: z.literal(true) }) } },
    },
    async ({ state }) => {
      const sub = state.user?.sub;
      if (!sub) throw new UnauthorizedError();
      const { email, link } = await issueVerificationLink(sub); // your auth provider
      // One send per user per minute, however often the button is clicked.
      await enqueueEmail(`verify-${sub}-${Math.floor(Date.now() / 60_000)}`, {
        to: email, // from the account, never from the request
        subject: "Confirm your email address",
        text: `Confirm your address: ${link}`,
        html: `<p><a href="${escapeHtml(link)}">Confirm your email address</a></p>`,
      });
      return { status: 202, body: { ok: true as const } };
    },
  );
}

declare function issueVerificationLink(userId: string): Promise<{ email: string; link: string }>;

import { app as azure } from "@azure/functions";
import { sender } from "../email/acs.ts";
import type { QueuedSend } from "../email/queue.ts";

azure.storageQueue<QueuedSend>("emailSend", {
  queueName: "email-send",
  connection: "EmailQueue", // reads EmailQueue__queueServiceUri, authenticates with the managed identity
  handler: async (job, context) => {
    if ((await alreadySent(job.key)) || (await isUndeliverable(job.msg.to))) return;
    try {
      await sender.send({ ...job.msg, idempotencyKey: job.operationId });
    } catch (err) {
      if ((err as { statusCode?: number }).statusCode === 400) {
        context.error(`ACS rejected ${job.key}`); // a bad request will not get better; no provider text
        return;
      }
      throw err; // 429, 5xx, network: the host retries, then moves the message to email-send-poison
    }
    await recordSent(job.key, job.operationId);
  },
});

// Your data layer: a table with a unique key column, plus the overview's undeliverable flag.
declare function alreadySent(key: string): Promise<boolean>;
declare function recordSent(key: string, operationId: string): Promise<void>;
declare function isUndeliverable(email: string): Promise<boolean>;

const pdf = await readInvoicePdf(invoice.id); // Uint8Array, from Blob Storage
await acs.beginSend({
  senderAddress: config.ACS_SENDER,
  content: { subject: `Invoice ${invoice.number}`, plainText: "Your invoice is attached." },
  recipients: { to: [{ address: invoice.customerEmail }] }, // looked up server-side
  attachments: [
    { name: `invoice-${invoice.number}.pdf`, contentType: "application/pdf", contentInBase64: Buffer.from(pdf).toString("base64") },
  ],
});

import { app as azure, type EventGridEvent } from "@azure/functions";
import { jobIdempotencyKey } from "@daloyjs/core";

azure.eventGrid("acsEmailEvents", {
  handler: async (event: EventGridEvent, context) => {
    if (event.eventType !== "Microsoft.Communication.EmailDeliveryReportReceived") return;
    const status = event.data?.status;
    const recipient = event.data?.recipient;
    if ((status === "Bounced" || status === "Suppressed") && typeof recipient === "string") {
      // Event Grid delivers at least once; a redelivered event keeps its id, so the insert is a no-op.
      const key = jobIdempotencyKey({ name: "email.suppress", key: `acs-${event.id}` });
      await recordSuppression(key, recipient.toLowerCase(), status);
    } else if (status === "FilteredSpam" || status === "Quarantined") {
      context.warn(`ACS ${String(status)} for message ${String(event.data?.messageId)}`); // no address in logs
    }
  },
});

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

import { z } from "zod";
import { App, BadRequestError, ForbiddenError, every, jobIdempotencyKey, jwk, readBodyLimited, requireRoles, safeJsonParse } from "@daloyjs/core";
import { enqueueSuppression } from "../email/queue.ts"; // same pattern as enqueueEmail, queue "email-suppress"
import { config } from "../config.ts";

interface EventGridEnvelope {
  id: string;
  eventType: string;
  data?: { validationCode?: unknown; status?: unknown; recipient?: unknown };
}

export function acsWebhookRoutes(app: App) {
  app.post(
    "/webhooks/acs-email",
    {
      operationId: "acsEmailEvents",
      hooks: every(
        jwk({
          jwks: `https://login.microsoftonline.com/${config.ENTRA_TENANT_ID}/discovery/v2.0/keys`,
          algorithms: ["RS256"],
          issuer: config.EVENTGRID_TOKEN_ISSUER,
          audience: config.EVENTGRID_WEBHOOK_AUDIENCE, // the webhook app's client id or App ID URI
          maxLifetimeSeconds: 86_400,
        }),
        requireRoles(["AzureEventGridSecureWebhookSubscriber"]),
      ),
      responses: {
        200: {
          description: "Accepted or validated",
          body: z.union([z.object({ ok: z.literal(true) }), z.object({ validationResponse: z.string() })]),
        },
        400: { description: "Malformed" },
        403: { description: "Unknown subscription" },
      },
    },
    async ({ request }) => {
      const subscription = (request.headers.get("aeg-subscription-name") ?? "").toLowerCase();
      if (subscription !== config.EVENTGRID_SUBSCRIPTION.toLowerCase()) throw new ForbiddenError();

      // No body schema: read bounded bytes, parse prototype-pollution safe.
      const raw = new TextDecoder().decode(await readBodyLimited(request, 1_048_576));
      const events = safeJsonParse(raw);
      if (!Array.isArray(events)) throw new BadRequestError();

      if (request.headers.get("aeg-event-type") === "SubscriptionValidation") {
        const first = events[0] as EventGridEnvelope | undefined;
        const code = first?.eventType === "Microsoft.EventGrid.SubscriptionValidationEvent" ? first.data?.validationCode : undefined;
        if (typeof code !== "string") throw new BadRequestError();
        return { status: 200, body: { validationResponse: code } };
      }

      for (const event of events as EventGridEnvelope[]) {
        if (event.eventType !== "Microsoft.Communication.EmailDeliveryReportReceived") continue;
        const { status, recipient } = event.data ?? {};
        if ((status === "Bounced" || status === "Suppressed") && typeof recipient === "string") {
          const key = jobIdempotencyKey({ name: "email.suppress", key: `acs-${event.id}` });
          await enqueueSuppression(key, recipient.toLowerCase(), status);
        }
      }
      return { status: 200, body: { ok: true as const } };
    },
  );
}