feat: store email attachments in R2 and expose as RSS enclosures
Attachments from incoming emails are uploaded to an optional Cloudflare R2
bucket and exposed as <enclosure> elements in RSS and <link rel="enclosure">
in Atom feeds, served at /files/{id}/{filename} with immutable caching.
R2 is opt-in: if ATTACHMENT_BUCKET is not bound the feature is a no-op.
Attachments are cleaned up from R2 on email/feed deletion and during
size-based feed trimming. Adds MockR2 to the test setup.
Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
This commit is contained in:
1 parent
3e28246c61
commit
e93bbb8d3e
15 files changed
+615
-19
No files matched your search
@@ -1,6 +1,6 @@
|
||||
import PostalMime from "postal-mime";
|
||||
import { Env } from "../types";
|
||||
import { processEmail } from "./email-processor";
|
||||
import { processEmail, RawAttachment } from "./email-processor";
|
||||
|
||||
export async function handleCloudflareEmail(
|
||||
message: ForwardableEmailMessage,
|
||||
@@ -21,6 +21,14 @@ export async function handleCloudflareEmail(
|
||||
headers[h.key] = h.value;
|
||||
}
|
||||
|
||||
const rawAttachments: RawAttachment[] = (email.attachments ?? [])
|
||||
.filter((a) => a.content instanceof ArrayBuffer)
|
||||
.map((a) => ({
|
||||
filename: a.filename || "attachment",
|
||||
contentType: a.mimeType || "application/octet-stream",
|
||||
content: a.content as ArrayBuffer,
|
||||
}));
|
||||
|
||||
await processEmail(
|
||||
{
|
||||
toAddress: message.to,
|
||||
@@ -30,6 +38,7 @@ export async function handleCloudflareEmail(
|
||||
content: email.html ?? email.text ?? "",
|
||||
receivedAt: email.date ? new Date(email.date).getTime() : Date.now(),
|
||||
headers,
|
||||
attachments: rawAttachments,
|
||||
},
|
||||
env,
|
||||
);
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { describe, it, expect, beforeEach } from "vitest";
|
||||
import "../test/setup";
|
||||
import { createMockEnv } from "../test/setup";
|
||||
import { processEmail, ProcessEmailInput } from "./email-processor";
|
||||
import { createMockEnv, MockR2 } from "../test/setup";
|
||||
import { processEmail, ProcessEmailInput, RawAttachment } from "./email-processor";
|
||||
|
||||
const VALID_FEED_ID = "apple.mountain.42";
|
||||
const VALID_TO = `${VALID_FEED_ID}@test.getmynews.app`;
|
||||
@@ -179,3 +179,126 @@ describe("processEmail", () => {
|
||||
expect(metadata.emails).toHaveLength(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("processEmail — attachments", () => {
|
||||
const pdfContent = new TextEncoder().encode("PDF bytes").buffer as ArrayBuffer;
|
||||
|
||||
const pdfAttachment: RawAttachment = {
|
||||
filename: "report.pdf",
|
||||
contentType: "application/pdf",
|
||||
content: pdfContent,
|
||||
};
|
||||
|
||||
it("skips R2 upload when ATTACHMENT_BUCKET is not configured", async () => {
|
||||
const env = createMockEnv();
|
||||
await env.EMAIL_STORAGE.put(
|
||||
`feed:${VALID_FEED_ID}:config`,
|
||||
JSON.stringify({}),
|
||||
);
|
||||
const res = await processEmail(
|
||||
makeInput({ attachments: [pdfAttachment] }),
|
||||
env as any,
|
||||
);
|
||||
expect(res.status).toBe(200);
|
||||
|
||||
const metadata = await env.EMAIL_STORAGE.get(
|
||||
`feed:${VALID_FEED_ID}:metadata`,
|
||||
"json",
|
||||
);
|
||||
const emailData = await env.EMAIL_STORAGE.get(metadata.emails[0].key, "json");
|
||||
expect(emailData.attachments).toBeUndefined();
|
||||
});
|
||||
|
||||
it("uploads attachments to R2 and stores AttachmentData in emailData", async () => {
|
||||
const env = createMockEnv({ withR2: true });
|
||||
const mockR2 = (env as any).ATTACHMENT_BUCKET as unknown as MockR2;
|
||||
await env.EMAIL_STORAGE.put(
|
||||
`feed:${VALID_FEED_ID}:config`,
|
||||
JSON.stringify({}),
|
||||
);
|
||||
const res = await processEmail(
|
||||
makeInput({ attachments: [pdfAttachment] }),
|
||||
env as any,
|
||||
);
|
||||
expect(res.status).toBe(200);
|
||||
|
||||
const metadata = await env.EMAIL_STORAGE.get(
|
||||
`feed:${VALID_FEED_ID}:metadata`,
|
||||
"json",
|
||||
);
|
||||
const emailData = await env.EMAIL_STORAGE.get(metadata.emails[0].key, "json");
|
||||
|
||||
expect(emailData.attachments).toHaveLength(1);
|
||||
expect(emailData.attachments[0].filename).toBe("report.pdf");
|
||||
expect(emailData.attachments[0].contentType).toBe("application/pdf");
|
||||
expect(emailData.attachments[0].size).toBe(pdfContent.byteLength);
|
||||
|
||||
const id = emailData.attachments[0].id;
|
||||
expect(mockR2._has(id)).toBe(true);
|
||||
});
|
||||
|
||||
it("stores attachmentIds in EmailMetadata for trim-time cleanup", async () => {
|
||||
const env = createMockEnv({ withR2: true });
|
||||
await env.EMAIL_STORAGE.put(
|
||||
`feed:${VALID_FEED_ID}:config`,
|
||||
JSON.stringify({}),
|
||||
);
|
||||
await processEmail(makeInput({ attachments: [pdfAttachment] }), env as any);
|
||||
|
||||
const metadata = await env.EMAIL_STORAGE.get(
|
||||
`feed:${VALID_FEED_ID}:metadata`,
|
||||
"json",
|
||||
);
|
||||
expect(metadata.emails[0].attachmentIds).toHaveLength(1);
|
||||
expect(typeof metadata.emails[0].attachmentIds[0]).toBe("string");
|
||||
});
|
||||
|
||||
it("deletes R2 objects when a trimmed email had attachments", async () => {
|
||||
const env = createMockEnv({ withR2: true });
|
||||
const mockR2 = (env as any).ATTACHMENT_BUCKET as unknown as MockR2;
|
||||
await env.EMAIL_STORAGE.put(
|
||||
`feed:${VALID_FEED_ID}:config`,
|
||||
JSON.stringify({}),
|
||||
);
|
||||
|
||||
// Store an old email with attachment in KV and metadata
|
||||
const oldKey = `feed:${VALID_FEED_ID}:111`;
|
||||
const oldAttachmentId = "old-attachment-uuid";
|
||||
const bigContent = "x".repeat(200);
|
||||
const oldEmail = JSON.stringify({
|
||||
subject: "Old",
|
||||
from: "[email protected]",
|
||||
content: bigContent,
|
||||
receivedAt: 111,
|
||||
headers: {},
|
||||
attachments: [{ id: oldAttachmentId, filename: "old.pdf", contentType: "application/pdf", size: 100 }],
|
||||
});
|
||||
await env.EMAIL_STORAGE.put(oldKey, oldEmail);
|
||||
|
||||
// Also put the attachment in mock R2
|
||||
await mockR2.put(oldAttachmentId, new ArrayBuffer(100));
|
||||
|
||||
await env.EMAIL_STORAGE.put(
|
||||
`feed:${VALID_FEED_ID}:metadata`,
|
||||
JSON.stringify({
|
||||
emails: [
|
||||
{
|
||||
key: oldKey,
|
||||
subject: "Old",
|
||||
receivedAt: 111,
|
||||
size: oldEmail.length,
|
||||
attachmentIds: [oldAttachmentId],
|
||||
},
|
||||
],
|
||||
}),
|
||||
);
|
||||
|
||||
// Process with tight size budget to force trimming
|
||||
const tinyEnv = { ...env, FEED_MAX_SIZE_BYTES: "50" };
|
||||
const res = await processEmail(makeInput({ subject: "New" }), tinyEnv as any);
|
||||
expect(res.status).toBe(200);
|
||||
|
||||
// Old attachment should be deleted from R2
|
||||
expect(mockR2._has(oldAttachmentId)).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -1,5 +1,11 @@
|
||||
import { EmailParser } from "../utils/email-parser";
|
||||
import { Env, FeedConfig, FeedMetadata } from "../types";
|
||||
import { AttachmentData, EmailMetadata, Env, FeedConfig, FeedMetadata } from "../types";
|
||||
|
||||
export interface RawAttachment {
|
||||
filename: string;
|
||||
contentType: string;
|
||||
content: ArrayBuffer;
|
||||
}
|
||||
|
||||
export interface ProcessEmailInput {
|
||||
toAddress: string;
|
||||
@@ -9,6 +15,7 @@ export interface ProcessEmailInput {
|
||||
content: string;
|
||||
receivedAt: number;
|
||||
headers?: Record<string, string>;
|
||||
attachments?: RawAttachment[];
|
||||
}
|
||||
|
||||
function normalizeEmail(value: string): string {
|
||||
@@ -34,6 +41,29 @@ function senderMatchesAllowlist(
|
||||
return senderDomain === normalizedDomain;
|
||||
}
|
||||
|
||||
async function uploadAttachments(
|
||||
attachments: RawAttachment[],
|
||||
bucket: R2Bucket,
|
||||
): Promise<AttachmentData[]> {
|
||||
return Promise.all(
|
||||
attachments.map(async (att) => {
|
||||
const id = crypto.randomUUID();
|
||||
await bucket.put(id, att.content, {
|
||||
httpMetadata: {
|
||||
contentType: att.contentType,
|
||||
contentDisposition: `attachment; filename="${att.filename}"`,
|
||||
},
|
||||
});
|
||||
return {
|
||||
id,
|
||||
filename: att.filename,
|
||||
contentType: att.contentType,
|
||||
size: att.content.byteLength,
|
||||
};
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
export async function processEmail(
|
||||
input: ProcessEmailInput,
|
||||
env: Env,
|
||||
@@ -74,12 +104,18 @@ export async function processEmail(
|
||||
}
|
||||
}
|
||||
|
||||
const storedAttachments: AttachmentData[] =
|
||||
env.ATTACHMENT_BUCKET && input.attachments?.length
|
||||
? await uploadAttachments(input.attachments, env.ATTACHMENT_BUCKET)
|
||||
: [];
|
||||
|
||||
const emailData = {
|
||||
subject: input.subject,
|
||||
from: input.from,
|
||||
content: input.content,
|
||||
receivedAt: input.receivedAt,
|
||||
headers: input.headers ?? {},
|
||||
...(storedAttachments.length > 0 ? { attachments: storedAttachments } : {}),
|
||||
};
|
||||
|
||||
const emailKey = `feed:${feedId}:${Date.now()}`;
|
||||
@@ -104,24 +140,36 @@ export async function processEmail(
|
||||
|
||||
const serialised = JSON.stringify(emailData);
|
||||
const serialisedSize = new TextEncoder().encode(serialised).byteLength;
|
||||
feedMetadata.emails.unshift({
|
||||
const newEntry: EmailMetadata = {
|
||||
key: emailKey,
|
||||
subject: emailData.subject,
|
||||
receivedAt: emailData.receivedAt,
|
||||
size: serialisedSize,
|
||||
});
|
||||
...(storedAttachments.length > 0
|
||||
? { attachmentIds: storedAttachments.map((a) => a.id) }
|
||||
: {}),
|
||||
};
|
||||
feedMetadata.emails.unshift(newEntry);
|
||||
|
||||
let totalSize = feedMetadata.emails.reduce((sum, e) => sum + (e.size ?? 0), 0);
|
||||
const toDelete: string[] = [];
|
||||
const toDelete: EmailMetadata[] = [];
|
||||
while (totalSize > maxBytes && feedMetadata.emails.length > 1) {
|
||||
const dropped = feedMetadata.emails.pop()!;
|
||||
totalSize -= dropped.size ?? 0;
|
||||
toDelete.push(dropped.key);
|
||||
toDelete.push(dropped);
|
||||
}
|
||||
|
||||
const r2Deletions =
|
||||
env.ATTACHMENT_BUCKET && toDelete.length > 0
|
||||
? toDelete
|
||||
.flatMap((e) => e.attachmentIds ?? [])
|
||||
.map((id) => env.ATTACHMENT_BUCKET!.delete(id))
|
||||
: [];
|
||||
|
||||
await Promise.all([
|
||||
env.EMAIL_STORAGE.put(feedMetadataKey, JSON.stringify(feedMetadata)),
|
||||
...toDelete.map((k) => env.EMAIL_STORAGE.delete(k)),
|
||||
...toDelete.map((e) => env.EMAIL_STORAGE.delete(e.key)),
|
||||
...r2Deletions,
|
||||
]);
|
||||
|
||||
console.log(`Successfully processed email for feed ${feedId}`);
|
||||
|
||||
+37
-2
@@ -1,6 +1,16 @@
|
||||
import { EmailParser } from "../utils/email-parser";
|
||||
import { Env } from "../types";
|
||||
import { processEmail } from "./email-processor";
|
||||
import { processEmail, RawAttachment } from "./email-processor";
|
||||
|
||||
export interface ForwardEmailAttachment {
|
||||
filename?: string;
|
||||
contentType?: string;
|
||||
size?: number;
|
||||
content?:
|
||||
| { type: "Buffer"; data: number[] }
|
||||
| ArrayBuffer
|
||||
| ArrayBufferView;
|
||||
}
|
||||
|
||||
export interface ForwardEmailPayload {
|
||||
recipients?: string[];
|
||||
@@ -17,7 +27,7 @@ export interface ForwardEmailPayload {
|
||||
headerLines?: Array<{ key: string; line: string }>;
|
||||
headers?: string;
|
||||
raw?: string;
|
||||
attachments?: Array<any>;
|
||||
attachments?: ForwardEmailAttachment[];
|
||||
}
|
||||
|
||||
function normalizeEmail(value: string): string {
|
||||
@@ -41,12 +51,36 @@ function extractSenderAddresses(payload: ForwardEmailPayload): string[] {
|
||||
return Array.from(new Set(matches.map(normalizeEmail)));
|
||||
}
|
||||
|
||||
function toArrayBuffer(
|
||||
content: ForwardEmailAttachment["content"],
|
||||
): ArrayBuffer | null {
|
||||
if (!content) return null;
|
||||
if (content instanceof ArrayBuffer) return content;
|
||||
if (ArrayBuffer.isView(content)) return (content as ArrayBufferView).buffer as ArrayBuffer;
|
||||
if (typeof content === "object" && content.type === "Buffer" && Array.isArray(content.data)) {
|
||||
return Uint8Array.from(content.data).buffer as ArrayBuffer;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export async function handleForwardEmail(
|
||||
payload: ForwardEmailPayload,
|
||||
env: Env,
|
||||
): Promise<Response> {
|
||||
const emailData = EmailParser.parseForwardEmailPayload(payload);
|
||||
|
||||
const rawAttachments: RawAttachment[] = (payload.attachments ?? [])
|
||||
.map((a) => {
|
||||
const buffer = toArrayBuffer(a.content);
|
||||
if (!buffer) return null;
|
||||
return {
|
||||
filename: a.filename || "attachment",
|
||||
contentType: a.contentType || "application/octet-stream",
|
||||
content: buffer,
|
||||
};
|
||||
})
|
||||
.filter((a): a is RawAttachment => a !== null);
|
||||
|
||||
return processEmail(
|
||||
{
|
||||
toAddress: payload.recipients?.[0] || "",
|
||||
@@ -56,6 +90,7 @@ export async function handleForwardEmail(
|
||||
content: emailData.content,
|
||||
receivedAt: emailData.receivedAt,
|
||||
headers: emailData.headers,
|
||||
attachments: rawAttachments,
|
||||
},
|
||||
env,
|
||||
);
|
||||
|
||||
Reference in new issue
Block a user