feat(cloudflare-module): support cloudflare:email and cloudflare:queue hooks

src/presets/cloudflare/runtime/cloudflare-module.ts

@@ -1,7 +1,7 @@
 import "#internal/nitro/virtual/polyfill";
 import { requestHasBody } from "#internal/nitro/utils";
 import { nitroApp } from "#internal/nitro/app";
-import { runCronTasks, useRuntimeConfig } from "#internal/nitro";
+import { runCronTasks, runEmailTask, useRuntimeConfig } from "#internal/nitro";
 import { getPublicAssetMeta } from "#internal/nitro/virtual/public-assets";
 
 import { withoutBase } from "ufo";
@@ -96,6 +96,26 @@ export default {
       );
     }
   },
+  email(event: any, env: CFModuleEnv, context: ExecutionContext) {
+    if (import.meta._tasks) {
+      (globalThis as any).__env__ = env;
+
+      const domain = (event.to as string).split("@")[1];
+
+      context.waitUntil(
+        runEmailTask(domain, {
+          context: {
+            ...event,
+            cloudflare: {
+              env,
+              context,
+            },
+          },
+          payload: {},
+        })
+      );
+    }
+  },
 };
 
 function assetsCacheControl(_request: Request) {

src/runtime/emailtask.ts

@@ -0,0 +1,117 @@
+import { emailTasks, scheduledEmailTasks } from "#internal/nitro/virtual/tasks";
+import { createError } from "h3";
+
+type MaybePromise<T> = T | Promise<T>;
+
+/** @experimental */
+export interface EmailContext {
+  readonly from?: string;
+  readonly to?: string;
+  readonly headers?: Headers;
+  readonly raw?: ReadableStream;
+  readonly rawSize?: number;
+
+  setReject?(reason: string): void;
+  forward?(rcptTo: string, headers?: Headers): Promise<void>;
+  reply?(message: EmailContext): Promise<void>;
+}
+
+/** @experimental */
+export interface EmailPayload {
+  [key: string]: unknown;
+}
+
+/** @experimental */
+export interface EmailMeta {
+  domain?: string;
+  description?: string;
+}
+
+/** @experimental */
+export interface EmailTaskEvent {
+  domain: string;
+  payload: EmailPayload;
+  context: EmailContext;
+}
+
+/** @experimental */
+export interface EmailTaskResult<RT = unknown> {
+  result?: RT;
+}
+
+/** @experimental */
+export interface EmailTask<RT = unknown> {
+  meta: EmailMeta;
+  run(event: EmailTaskEvent): MaybePromise<{ result?: RT }>;
+}
+
+/** @experimental */
+export function defineEmailTask<RT = unknown>(
+  def: EmailTask<RT>
+): EmailTask<RT> {
+  if (typeof def.run !== "function") {
+    def.run = () => {
+      throw new TypeError("Task must implement a `run` method!");
+    };
+  }
+  return def;
+}
+
+const __runningEmailTasks__: {
+  [name: string]: MaybePromise<EmailTaskResult<any>>;
+} = {};
+
+/** @experimental */
+export async function runEmailTask<RT = unknown>(
+  domain: string,
+  {
+    payload = {},
+    context = {},
+  }: { payload?: EmailPayload; context?: EmailContext } = {}
+): Promise<EmailTaskResult<RT>> {
+  if (__runningEmailTasks__[domain]) {
+    return __runningEmailTasks__[domain];
+  }
+
+  if (!(domain in emailTasks)) {
+    throw createError({
+      message: `Task \`${domain}\` is not available!`,
+      statusCode: 404,
+    });
+  }
+
+  if (!emailTasks[domain].resolve) {
+    throw createError({
+      message: `Task \`${domain}\` is not implemented!`,
+      statusCode: 501,
+    });
+  }
+
+  const handler = (await emailTasks[domain].resolve!()) as EmailTask<RT>;
+  const taskEvent: EmailTaskEvent = { domain, payload, context };
+  __runningEmailTasks__[domain] = handler.run(taskEvent);
+
+  try {
+    const res = await __runningEmailTasks__[domain];
+    return res;
+  } finally {
+    delete __runningEmailTasks__[domain];
+  }
+}
+
+/** @experimental */
+export function getEmailTasks(domain: string): string[] {
+  return (
+    (scheduledEmailTasks || []).find((task) => task.domain.endsWith(domain))
+      ?.tasks || []
+  );
+}
+
+export function runEmailTasks(
+  domain: string,
+  ctx: { payload?: EmailPayload; context?: EmailContext }
+): Promise<EmailTaskResult[]> {
+  return Promise.all(
+    getEmailTasks(domain).map((name) => runEmailTask(name, ctx))
+  );
+}

src/runtime/index.ts

@@ -3,6 +3,7 @@ export * from "./cache";
 export { useNitroApp } from "./app";
 export * from "./plugin";
 export * from "./task";
+export * from "./emailtask";
 export * from "./renderer";
 export { getRouteRules, getRouteRulesForPath } from "./route-rules";
 export { defineRouteMeta } from "./meta";

src/runtime/virtual/tasks.ts

@@ -1,8 +1,18 @@
 import type { Task, TaskMeta } from "../task";
+import { EmailMeta, EmailTask } from "#internal/nitro/emailtask";
 
 export const tasks: Record<
   string,
   { resolve?: () => Promise<Task>; meta: TaskMeta }
 > = {};
 
 export const scheduledTasks: false | { cron: string; tasks: string[] }[] = [];
+
+export const emailTasks: Record<
+  string,
+  { resolve?: () => Promise<EmailTask>; meta: EmailMeta }
+> = {};
+
+export const scheduledEmailTasks:
+  | false
+  | { domain: string; tasks: string[] }[] = [];