邮件存储与处理

使用全面的存储、队列处理和支持工单自动化来存储和处理传入邮件，实现更高效的邮件工作流管理。

将邮件存储到 KV

将邮件存储在 KV 命名空间中，供后续处理。本示例使用 mimetext ↗ 构建回复，这需要 nodejs_compat 兼容性标志。

import { EmailMessage } from "cloudflare:email";
import { createMimeMessage } from "mimetext";

interface Env {
	EMAIL: SendEmail;
	EMAILS: KVNamespace;
	SUPPORT_TICKETS: KVNamespace;
}

export default {
	async email(message, env, ctx): Promise<void> {
		const emailId = `email-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;

		// Read email content as an ArrayBuffer, then decode to a string
		const rawBuffer = await new Response(message.raw).arrayBuffer();
		const rawContent = new TextDecoder().decode(rawBuffer);

		// Store email metadata and content
		const emailData = {
			id: emailId,
			from: message.from,
			to: message.to,
			subject: message.headers.get("subject"),
			timestamp: new Date().toISOString(),
			size: message.rawSize,
			rawContent: rawContent,
			processed: false,
		};

		await env.EMAILS.put(emailId, JSON.stringify(emailData));

		// Process based on recipient
		if (message.to.includes("support@")) {
			await handleSupportEmail(message, env, emailId);
		} else {
			await message.forward("[email protected]");
		}
	},
};

async function handleSupportEmail(message, env, emailId) {
	const ticketId = `TICKET-${Date.now()}`;

	// Create support ticket
	const ticketData = {
		id: ticketId,
		emailId: emailId,
		from: message.from,
		subject: message.headers.get("subject"),
		status: "open",
		priority: "normal",
		createdAt: new Date().toISOString(),
		updatedAt: new Date().toISOString(),
	};

	await env.SUPPORT_TICKETS.put(ticketId, JSON.stringify(ticketData));

	// Send threaded auto-reply with ticket number
	const messageId = message.headers.get("Message-ID");
	const reply = createMimeMessage();
	if (messageId) {
		reply.setHeader("In-Reply-To", messageId);
		reply.setHeader("References", messageId);
	}
	reply.setSender(message.to);
	reply.setRecipient(message.from);
	reply.setSubject(`Support Ticket Created: ${ticketId}`);
	reply.addMessage({
		contentType: "text/html",
		data: `
            <h1>Support Ticket Created</h1>
            <p>Your support request has been received and assigned ticket number: <strong>${ticketId}</strong></p>
            <p>We will respond within 2-4 hours during business hours.</p>
            <hr>
            <p><em>Original subject: ${message.headers.get("subject")}</em></p>
        `,
	});

	await message.reply(
		new EmailMessage(message.to, message.from, reply.asRaw()),
	);

	// Forward to support team
	await message.forward("[email protected]");
}

基于队列的处理

使用 Cloudflare Queues 异步处理邮件。email() 处理程序通过 message.reply() 发送带线程的确认，并将消息入队以便后续处理。队列消费者在单独的调用中运行，此时原始 EmailMessage 已不可用，因此它发送的任何后续邮件都使用 env.EMAIL.send()。

import { EmailMessage } from "cloudflare:email";
import { createMimeMessage } from "mimetext";

interface Env {
	EMAIL: SendEmail;
	EMAIL_QUEUE: Queue;
	EMAIL_STORAGE: KVNamespace;
	EMAIL_ANALYTICS: AnalyticsEngine;
	SUPPORT_TICKETS?: KVNamespace;
	SALES_LEADS?: KVNamespace;
}

interface EmailQueueMessage {
	emailId: string;
	from: string;
	to: string;
	subject: string;
	timestamp: string;
	priority: "low" | "normal" | "high" | "urgent";
	category: string;
}

export default {
	// Handle incoming emails
	async email(message, env, ctx): Promise<void> {
		const emailId = `email-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;

		// Read raw email content as an ArrayBuffer, then decode to a string
		const rawBuffer = await new Response(message.raw).arrayBuffer();
		const rawContent = new TextDecoder().decode(rawBuffer);

		// Store email with metadata
		const emailData = {
			id: emailId,
			from: message.from,
			to: message.to,
			subject: message.headers.get("subject"),
			timestamp: new Date().toISOString(),
			size: message.rawSize,
			rawContent: rawContent,
			processed: false,
			status: "queued",
		};

		await env.EMAIL_STORAGE.put(emailId, JSON.stringify(emailData));

		// Determine priority and category
		const priority = determinePriority(message);
		const category = determineCategory(message);

		// Queue email for processing
		const queueMessage: EmailQueueMessage = {
			emailId,
			from: message.from,
			to: message.to,
			subject: message.headers.get("subject") || "",
			timestamp: new Date().toISOString(),
			priority,
			category,
		};

		await env.EMAIL_QUEUE.send(queueMessage, {
			delaySeconds: priority === "urgent" ? 0 : priority === "high" ? 5 : 30,
		});

		// Send threaded immediate auto-reply
		const messageId = message.headers.get("Message-ID");
		const reply = createMimeMessage();
		if (messageId) {
			reply.setHeader("In-Reply-To", messageId);
			reply.setHeader("References", messageId);
		}
		reply.setSender(message.to);
		reply.setRecipient(message.from);
		reply.setSubject(`Re: ${message.headers.get("subject")}`);
		reply.addMessage({
			contentType: "text/plain",
			data: "Thank you for your message. It has been queued for processing.",
		});

		await message.reply(
			new EmailMessage(message.to, message.from, reply.asRaw()),
		);
	},

	// Process queued emails
	async queue(batch, env, ctx): Promise<void> {
		console.log(`Processing ${batch.messages.length} queued emails`);

		for (const message of batch.messages) {
			try {
				const emailData = message.body as EmailQueueMessage;

				console.log(
					`Processing ${emailData.category} email from ${emailData.from}`,
				);

				// Get stored email content
				const storedEmailData = await env.EMAIL_STORAGE.get(emailData.emailId);
				if (!storedEmailData) {
					console.error(`Email data not found: ${emailData.emailId}`);
					message.ack();
					continue;
				}

				const emailContent = JSON.parse(storedEmailData);

				// Process based on category
				let processResult;
				switch (emailData.category) {
					case "support":
						processResult = await processSupport(emailData, emailContent, env);
						break;
					case "sales":
						processResult = await processSales(emailData, emailContent, env);
						break;
					case "billing":
						processResult = await processBilling(emailData, emailContent, env);
						break;
					default:
						processResult = await processGeneral(emailData, emailContent, env);
				}

				// Update email status
				emailContent.processed = true;
				emailContent.status = "completed";
				emailContent.processedAt = new Date().toISOString();
				emailContent.processingResult = processResult;

				await env.EMAIL_STORAGE.put(
					emailData.emailId,
					JSON.stringify(emailContent),
				);

				// Track processing metrics
				env.EMAIL_ANALYTICS?.writeDataPoint({
					blobs: [
						"email_processed",
						emailData.from,
						emailData.to,
						emailData.category,
						emailData.priority,
					],
					doubles: [1, emailContent.size],
					indexes: [
						`category:${emailData.category}`,
						`priority:${emailData.priority}`,
					],
				});

				message.ack();
			} catch (error) {
				console.error("Failed to process email:", error);
				message.retry();
			}
		}
	},
};

function determinePriority(message): "low" | "normal" | "high" | "urgent" {
	const subject = (message.headers.get("subject") || "").toLowerCase();
	const to = message.to.toLowerCase();

	if (subject.includes("urgent") || subject.includes("emergency")) {
		return "urgent";
	}

	if (
		to.includes("support") &&
		(subject.includes("down") || subject.includes("error"))
	) {
		return "high";
	}

	if (to.includes("sales") || to.includes("billing")) {
		return "high";
	}

	return "normal";
}

function determineCategory(message): string {
	const to = message.to.toLowerCase();
	const subject = (message.headers.get("subject") || "").toLowerCase();

	if (
		to.includes("support") ||
		subject.includes("help") ||
		subject.includes("issue")
	) {
		return "support";
	}

	if (
		to.includes("sales") ||
		subject.includes("quote") ||
		subject.includes("pricing")
	) {
		return "sales";
	}

	if (
		to.includes("billing") ||
		subject.includes("invoice") ||
		subject.includes("payment")
	) {
		return "billing";
	}

	return "general";
}

async function processSupport(
	emailData: EmailQueueMessage,
	emailContent: any,
	env: Env,
) {
	const ticketId = `TICKET-${Date.now()}`;

	// Create support ticket
	const ticketData = {
		id: ticketId,
		emailId: emailData.emailId,
		from: emailData.from,
		subject: emailData.subject,
		priority: emailData.priority,
		status: "open",
		category: "support",
		createdAt: new Date().toISOString(),
		updatedAt: new Date().toISOString(),
		content: emailContent.rawContent.substring(0, 5000), // Limit stored content
	};

	await env.SUPPORT_TICKETS?.put(ticketId, JSON.stringify(ticketData));

	// Send confirmation email
	await env.EMAIL.send({
		to: emailData.from,
		from: "[email protected]",
		subject: `Support Ticket Created: ${ticketId}`,
		html: `
			<h2>Support Ticket Created</h2>
			<p>Your support request has been received and assigned ticket number: <strong>${ticketId}</strong></p>
			<p><strong>Priority:</strong> ${emailData.priority}</p>
			<p>We will respond based on the priority level:</p>
			<ul>
				<li><strong>Urgent:</strong> Within 1 hour</li>
				<li><strong>High:</strong> Within 4 hours</li>
				<li><strong>Normal:</strong> Within 24 hours</li>
			</ul>
			<hr>
			<p><em>Original subject: ${emailData.subject}</em></p>
		`,
	});

	return { ticketId, action: "ticket_created" };
}

async function processSales(
	emailData: EmailQueueMessage,
	emailContent: any,
	env: Env,
) {
	// Create sales lead
	const leadId = `LEAD-${Date.now()}`;

	const leadData = {
		id: leadId,
		emailId: emailData.emailId,
		contact: emailData.from,
		subject: emailData.subject,
		priority: emailData.priority,
		status: "new",
		source: "email",
		createdAt: new Date().toISOString(),
	};

	await env.SALES_LEADS?.put(leadId, JSON.stringify(leadData));

	// Send sales response
	await env.EMAIL.send({
		to: emailData.from,
		from: "[email protected]",
		subject: `Re: ${emailData.subject}`,
		html: `
			<h2>Thank you for your interest!</h2>
			<p>We've received your sales inquiry and assigned it reference: <strong>${leadId}</strong></p>
			<p>A member of our sales team will contact you within 24 hours.</p>
			<p>Best regards,<br>Sales Team</p>
		`,
	});

	return { leadId, action: "lead_created" };
}

async function processBilling(
	emailData: EmailQueueMessage,
	emailContent: any,
	env: Env,
) {
	// Handle billing inquiries
	await env.EMAIL.send({
		to: emailData.from,
		from: "[email protected]",
		subject: `Re: ${emailData.subject}`,
		html: `
			<h2>Billing Inquiry Received</h2>
			<p>Thank you for contacting our billing department.</p>
			<p>Your inquiry has been forwarded to our billing specialists who will respond within 2 business hours.</p>
			<p>For immediate assistance, please call: +1-800-555-0123</p>
		`,
	});

	return { action: "billing_forwarded" };
}

async function processGeneral(
	emailData: EmailQueueMessage,
	emailContent: any,
	env: Env,
) {
	// Handle general inquiries
	await env.EMAIL.send({
		to: emailData.from,
		from: "[email protected]",
		subject: `Re: ${emailData.subject}`,
		text: `
			Thank you for contacting us.
			
			We have received your message and will respond within 48 hours.
			
			For urgent matters, please contact our support team at [email protected].
			
			Best regards,
			Customer Service Team
		`,
	});

	return { action: "general_acknowledged" };
}

后续步骤

Email handler — email() 处理程序及其操作的参考文档。
Spam filtering — 在存储或处理之前拦截垃圾邮件。
Hard bounce handling — 检测并处理退信通知。