diff --git a/docs/contributing/architecture/primitives.yaml b/docs/contributing/architecture/primitives.yaml index f0ee703860..ae2641ca4c 100644 --- a/docs/contributing/architecture/primitives.yaml +++ b/docs/contributing/architecture/primitives.yaml @@ -497,7 +497,9 @@ primitives: name: Email blob storage (R2) summary: EMAIL_BLOBS R2 bucket holding raw email MIME under the frozen key contract - email-raw:v1:{userId}/{messageId}; D1 email_messages keeps metadata plus + email-raw:v1:{userId}/{messageId} and standalone outbound attachment bytes + under email-attachment:v1:{userId}/{messageId}/{attachmentId} + (storage_kind 'external'); D1 email_messages keeps metadata plus raw_mime_key, with inline raw_mime as the fallback when the binding is unavailable. Covered by account deletion and retention. code: diff --git a/docs/use/email-primitives.md b/docs/use/email-primitives.md index 00e0dbc4b3..56c7327faf 100644 --- a/docs/use/email-primitives.md +++ b/docs/use/email-primitives.md @@ -46,7 +46,11 @@ Use the MCP `email` domain: - `email_send` sends a notification email to your own account email address (notify-self only; any other recipient is rejected). - `email_reply` replies to a stored inbound message. The recipient always comes - from the stored message. + from the stored message. Optional `attachments` (up to 10 of + `{ filename, content_type, content_base64 }`) are sent with the reply and + stored as `external` attachments readable later via `email_attachment_get`. + With attachments, the whole message (bodies plus decoded attachment bytes) + must fit the plan's `email_message_bytes` per-message cap. - `email_attachment_get` returns stored attachment bytes by attachment id. - `email_message_list` lists stored inbound and outbound messages. - `email_message_search` searches stored messages by case-insensitive substring diff --git a/packages/shared/src/outbound-email.ts b/packages/shared/src/outbound-email.ts index f616a98e1e..404836914b 100644 --- a/packages/shared/src/outbound-email.ts +++ b/packages/shared/src/outbound-email.ts @@ -2,6 +2,7 @@ import { array, createSchema, fail, + literal, object, optional, string, @@ -36,6 +37,19 @@ const optionalHeadersSchema = createSchema< return { value: headers } }) +const outboundEmailAttachmentSchema = object({ + /** Base64-encoded bytes, as the Cloudflare Email REST API expects. */ + content: nonEmptyStringSchema, + filename: nonEmptyStringSchema, + type: nonEmptyStringSchema, + disposition: union([literal('attachment'), literal('inline')]), + contentId: optional(nonEmptyStringSchema), +}) + +export type OutboundEmailAttachment = InferOutput< + typeof outboundEmailAttachmentSchema +> + const outboundEmailSchema = object({ from: nonEmptyStringSchema, to: union([nonEmptyStringSchema, array(nonEmptyStringSchema)]), @@ -44,6 +58,7 @@ const outboundEmailSchema = object({ text: optional(nonEmptyStringSchema), replyTo: optional(nonEmptyStringSchema), headers: optionalHeadersSchema, + attachments: optional(array(outboundEmailAttachmentSchema)), }) export type OutboundEmail = InferOutput diff --git a/packages/worker/src/app/account-deletion.ts b/packages/worker/src/app/account-deletion.ts index 7276fd2ac8..26b38e2b8d 100644 --- a/packages/worker/src/app/account-deletion.ts +++ b/packages/worker/src/app/account-deletion.ts @@ -95,7 +95,7 @@ type UserDeletionInventory = { vectorIds: Array storageIds: Array bundleKvKeys: Array - emailRawMimeKeys: Array + emailBlobKeys: Array sourceSnapshots: Array savedPackages: Array repoSessions: Array @@ -283,15 +283,30 @@ async function listUserPackageServices(env: Env, userId: string) { })) } -async function listUserEmailRawMimeKeys(env: Env, userId: string) { - const rows = await env.APP_DB.prepare( - `SELECT raw_mime_key - FROM email_messages - WHERE user_id = ? AND raw_mime_key IS NOT NULL`, - ) - .bind(userId) - .all<{ raw_mime_key: string }>() - return uniqueStrings((rows.results ?? []).map((row) => row.raw_mime_key)) +async function listUserEmailBlobKeys(env: Env, userId: string) { + const [rawMimeRows, attachmentRows] = await Promise.all([ + env.APP_DB.prepare( + `SELECT raw_mime_key + FROM email_messages + WHERE user_id = ? AND raw_mime_key IS NOT NULL`, + ) + .bind(userId) + .all<{ raw_mime_key: string }>(), + // Externally stored attachments (outbound mail) have their own R2 + // objects, separate from any raw-MIME blob. + env.APP_DB.prepare( + `SELECT attachment.storage_key AS storage_key + FROM email_attachments attachment + JOIN email_messages message ON message.id = attachment.message_id + WHERE message.user_id = ? AND attachment.storage_key IS NOT NULL`, + ) + .bind(userId) + .all<{ storage_key: string }>(), + ]) + return uniqueStrings([ + ...(rawMimeRows.results ?? []).map((row) => row.raw_mime_key), + ...(attachmentRows.results ?? []).map((row) => row.storage_key), + ]) } async function listUserCommunityListingIds(env: Env, userId: string) { @@ -342,7 +357,7 @@ async function collectUserDeletionInventory(input: { const [ vectorIds, storageIds, - emailRawMimeKeys, + emailBlobKeys, sourceSnapshots, savedPackages, repoSessions, @@ -361,11 +376,9 @@ async function collectUserDeletionInventory(input: { input.warnings.push(`Failed to enumerate storage ids: ${message}`) return [] as Array }), - listUserEmailRawMimeKeys(input.env, input.userId).catch((error) => { - const message = error instanceof Error ? error.message : String(error) - input.warnings.push( - `Failed to enumerate email raw-MIME blob keys: ${message}`, - ) + listUserEmailBlobKeys(input.env, input.userId).catch((error) => { + const message = getErrorMessage(error) + input.warnings.push(`Failed to enumerate email blob keys: ${message}`) return [] as Array }), listUserSourceSnapshots(input.env, input.userId).catch((error) => { @@ -418,7 +431,7 @@ async function collectUserDeletionInventory(input: { vectorIds, storageIds, bundleKvKeys, - emailRawMimeKeys, + emailBlobKeys, sourceSnapshots, savedPackages, repoSessions, @@ -1068,7 +1081,7 @@ export async function deleteUserAccount(input: { result.deletedEmailBlobs = await deleteEmailBlobs({ blobs: input.env.EMAIL_BLOBS, - keys: inventory.emailRawMimeKeys, + keys: inventory.emailBlobKeys, warnings, }) diff --git a/packages/worker/src/app/email/cloudflare-email.ts b/packages/worker/src/app/email/cloudflare-email.ts index a6f2fce24e..8a08208e3c 100644 --- a/packages/worker/src/app/email/cloudflare-email.ts +++ b/packages/worker/src/app/email/cloudflare-email.ts @@ -130,13 +130,14 @@ async function sendViaCloudflareApi( export async function sendCloudflareEmail( config: CloudflareEmailClientConfig, - message: Omit & - Partial>, + message: Omit & + Partial>, ): Promise { const normalized = normalizeEmailPayload({ ...message, replyTo: message.replyTo, headers: message.headers, + attachments: message.attachments, }) const apiBaseUrl = typeof config.apiBaseUrl === 'string' && config.apiBaseUrl.trim().length > 0 diff --git a/packages/worker/src/app/retention.node.test.ts b/packages/worker/src/app/retention.node.test.ts index 8da07100c1..06802ea3cf 100644 --- a/packages/worker/src/app/retention.node.test.ts +++ b/packages/worker/src/app/retention.node.test.ts @@ -753,6 +753,16 @@ test('email message retention deletes old user rows, R2 blobs, attachments, and ) VALUES ('att-old', 'msg-old-blob', 'text/plain', 1, 'raw-mime', ?)`, ) .run(daysAgo(emailMessageRetentionDays + 1)) + // Externally stored attachment bytes (outbound mail) live in their own + // R2 object that must be deleted alongside the raw-MIME blobs. + sqlite + .prepare( + `INSERT INTO email_attachments ( + id, message_id, content_type, size, storage_kind, storage_key, created_at + ) VALUES ('att-ext', 'msg-old-plain', 'text/plain', 1, 'external', + 'email-attachment:v1:user-1/msg-old-plain/att-ext', ?)`, + ) + .run(daysAgo(emailMessageRetentionDays + 1)) const blobDelete = vi.fn(async () => undefined) const result = await pruneUserEmailMessagesForRetention({ @@ -763,14 +773,18 @@ test('email message retention deletes old user rows, R2 blobs, attachments, and expect(result).toEqual({ deletedMessages: 2, - deletedAttachments: 1, + deletedAttachments: 2, deletedThreads: 1, deletedRawMimeBlobs: 1, + deletedAttachmentBlobs: 1, blobDeleteErrors: 0, hasMore: false, nextCursor: null, }) - expect(blobDelete).toHaveBeenCalledWith(['raw-mime/user-1/msg-old-blob']) + expect(blobDelete).toHaveBeenCalledWith([ + 'raw-mime/user-1/msg-old-blob', + 'email-attachment:v1:user-1/msg-old-plain/att-ext', + ]) expect(idsForTable(sqlite, 'email_messages')).toEqual([ 'msg-boundary', 'msg-fresh', diff --git a/packages/worker/src/app/retention.ts b/packages/worker/src/app/retention.ts index 45265bb59a..683f5ed1de 100644 --- a/packages/worker/src/app/retention.ts +++ b/packages/worker/src/app/retention.ts @@ -26,6 +26,12 @@ export const retentionCronGateMinutes = 5 export const retentionCronIntervalMinutes = 60 export const retentionDefaultBatchSize = 250 export const retentionDeleteIdsMaxParameters = 100 +/** + * R2 bulk deletes accept up to 1000 keys per call. A 250-message batch with + * raw MIME plus up to ten external attachment keys each can exceed that, so + * blob deletes are chunked below this ceiling. + */ +export const retentionBlobDeleteMaxKeys = 1000 export const publishedBundleArtifactRetentionBatchSize = 100 /** * Total wall-clock budget for one retention cron run. Each hourly run keeps @@ -202,6 +208,7 @@ export type RetentionPruneResult = { deletedAttachments: number deletedThreads: number deletedRawMimeBlobs: number + deletedAttachmentBlobs: number blobDeleteErrors: number } entitlementDailyCounters: number @@ -714,6 +721,7 @@ export type EmailMessageRetentionResult = { deletedAttachments: number deletedThreads: number deletedRawMimeBlobs: number + deletedAttachmentBlobs: number blobDeleteErrors: number hasMore: boolean nextCursor: EmailMessageRetentionCursor | null @@ -760,27 +768,87 @@ export async function pruneUserEmailMessagesForRetention(input: { deletedAttachments: 0, deletedThreads: 0, deletedRawMimeBlobs: 0, + deletedAttachmentBlobs: 0, blobDeleteErrors: 0, hasMore: false, nextCursor: null, } - const rowsWithBlob = rows.filter((row) => row.raw_mime_key !== null) - let deletableRows = rows - if (rowsWithBlob.length > 0) { - // Never delete a row whose R2 blob could not be deleted first: once - // the row is gone its raw_mime_key is lost and the blob is orphaned - // forever. Failed rows are skipped and retried on a later run. + // Externally stored attachments (outbound mail) have their own R2 + // objects that must be deleted along with the raw-MIME blobs. + const attachmentKeysByMessageId = new Map>() + for ( + let index = 0; + index < rows.length; + index += retentionDeleteIdsMaxParameters + ) { + const chunk = rows.slice(index, index + retentionDeleteIdsMaxParameters) + const chunkPlaceholders = chunk.map(() => '?').join(', ') + const attachmentRows = await runD1WithRetry(() => + input.db + .prepare( + `SELECT message_id, storage_key + FROM email_attachments + WHERE storage_key IS NOT NULL AND message_id IN (${chunkPlaceholders})`, + ) + .bind(...chunk.map((row) => row.id)) + .all<{ message_id: string; storage_key: string }>(), + ) + for (const attachmentRow of attachmentRows.results ?? []) { + const keys = attachmentKeysByMessageId.get(attachmentRow.message_id) + if (keys) keys.push(attachmentRow.storage_key) + else { + attachmentKeysByMessageId.set(attachmentRow.message_id, [ + attachmentRow.storage_key, + ]) + } + } + } + const blobKeysForRow = (row: (typeof rows)[number]) => [ + ...(row.raw_mime_key !== null ? [row.raw_mime_key] : []), + ...(attachmentKeysByMessageId.get(row.id) ?? []), + ] + const rowsWithBlob = rows.filter((row) => blobKeysForRow(row).length > 0) + // Never delete a row whose R2 blobs could not be deleted first: once + // the row is gone its blob keys are lost and the blobs are orphaned + // forever. Failed rows are skipped and retried on a later run. Deletes + // are chunked below R2's bulk-delete key limit, keeping each message's + // keys inside one call so a failure skips whole messages. + const failedRowIds = new Set() + let chunkRows: Array<(typeof rows)[number]> = [] + let chunkKeys: Array = [] + const deleteChunk = async () => { + if (chunkKeys.length === 0) return try { - await input.blobs.delete( - rowsWithBlob.map((row) => row.raw_mime_key as string), + await input.blobs.delete(chunkKeys) + result.deletedRawMimeBlobs += chunkRows.filter( + (row) => row.raw_mime_key !== null, + ).length + result.deletedAttachmentBlobs += chunkRows.reduce( + (count, row) => + count + (attachmentKeysByMessageId.get(row.id)?.length ?? 0), + 0, ) - result.deletedRawMimeBlobs = rowsWithBlob.length } catch (error) { - result.blobDeleteErrors = rowsWithBlob.length - deletableRows = rows.filter((row) => row.raw_mime_key === null) - console.warn('retention-email-raw-mime-delete-failed', { error }) + result.blobDeleteErrors += chunkKeys.length + for (const row of chunkRows) failedRowIds.add(row.id) + console.warn('retention-email-blob-delete-failed', { error }) + } + chunkRows = [] + chunkKeys = [] + } + for (const row of rowsWithBlob) { + const keys = blobKeysForRow(row) + if ( + chunkKeys.length > 0 && + chunkKeys.length + keys.length > retentionBlobDeleteMaxKeys + ) { + await deleteChunk() } + chunkRows.push(row) + chunkKeys.push(...keys) } + await deleteChunk() + const deletableRows = rows.filter((row) => !failedRowIds.has(row.id)) const messageIds = deletableRows.map((row) => row.id) // email_attachments rows reference email_messages via FK, so delete the // dependent attachment rows first. @@ -930,6 +998,7 @@ export async function pruneRetention(input: { deletedAttachments: 0, deletedThreads: 0, deletedRawMimeBlobs: 0, + deletedAttachmentBlobs: 0, blobDeleteErrors: 0, }, entitlementDailyCounters: 0, @@ -1027,6 +1096,8 @@ export async function pruneRetention(input: { result.emailMessages.deletedAttachments += batch.deletedAttachments result.emailMessages.deletedThreads += batch.deletedThreads result.emailMessages.deletedRawMimeBlobs += batch.deletedRawMimeBlobs + result.emailMessages.deletedAttachmentBlobs += + batch.deletedAttachmentBlobs result.emailMessages.blobDeleteErrors += batch.blobDeleteErrors emailMessagesCursor = batch.nextCursor return batch.hasMore diff --git a/packages/worker/src/email/outbound.ts b/packages/worker/src/email/outbound.ts index de23e657e1..1746d3993e 100644 --- a/packages/worker/src/email/outbound.ts +++ b/packages/worker/src/email/outbound.ts @@ -1,9 +1,11 @@ +import { base64ToBytes } from '@kody-internal/shared/base64.ts' import { getErrorMessage } from '@kody-internal/shared/error-message.ts' import { sendCloudflareEmail } from '#app/email/cloudflare-email.ts' import { isAccountEmailVerified } from '#app/email-verification.ts' import { normalizeEmail } from '#app/normalize-email.ts' import { nullPlanEmailFallbackLimits } from '#worker/entitlements/plans.ts' import { + assertWithinEntitlement, assertWithinStorageBytesEntitlement, consumeDailyEntitlement, estimateEntitlementStorageEntryBytes, @@ -13,9 +15,11 @@ import { normalizeEmailAddress } from './address.ts' import { resolveUserPlatformSender } from './platform-address.ts' import { createEmailThread, + emailAttachmentBlobKey, ensurePlatformSenderIdentity, getEmailMessageById, getEmailMessageByMessageIdHeader, + insertEmailAttachments, insertEmailMessage, insertEmailDeliveryEvent, updateEmailMessageDelivery, @@ -35,6 +39,28 @@ type SendEmailEnv = Pick< | 'CLOUDFLARE_API_TOKEN' > +/** + * Ceiling on the number of files per outbound message. Total size is + * governed separately by the plan's email_message_bytes limit (and the + * provider's own 5 MiB hard cap far above it). + */ +export const maxOutboundEmailAttachments = 10 + +/** + * Hard ceiling on combined decoded attachment bytes, matching the + * provider's 5 MiB message cap. Enforced from base64 string lengths + * before any decoding so oversized inputs never allocate buffers; + * plan-level email_message_bytes still applies afterwards. + */ +export const maxOutboundEmailAttachmentTotalBytes = 5 * 1024 * 1024 + +export type OutboundEmailAttachmentInput = { + filename: string + contentType: string + /** Base64-encoded attachment bytes. */ + contentBase64: string +} + export type EmailSendInput = { env: SendEmailEnv userId: string @@ -53,6 +79,7 @@ export type EmailSendInput = { references?: Array threadId?: string | null inboxId?: string | null + attachments?: Array } & ( | { /** @@ -143,6 +170,126 @@ function buildProviderHeaders(headers: Record) { ) } +type PreparedOutboundAttachment = { + filename: string + contentType: string + bytes: Uint8Array + contentBase64: string +} + +function prepareOutboundAttachments( + attachments: Array | undefined, +): Array { + if (!attachments || attachments.length === 0) return [] + if (attachments.length > maxOutboundEmailAttachments) { + throw new Error( + `Outbound email supports at most ${maxOutboundEmailAttachments} attachments.`, + ) + } + // Base64 encodes 3 bytes per 4 characters, so the string length bounds + // the decoded size without allocating anything. + const estimatedTotalBytes = attachments.reduce( + (total, attachment) => + total + Math.floor((attachment.contentBase64.trim().length * 3) / 4), + 0, + ) + if (estimatedTotalBytes > maxOutboundEmailAttachmentTotalBytes) { + throw new Error( + `Outbound email attachments exceed the ${Math.floor(maxOutboundEmailAttachmentTotalBytes / (1024 * 1024))} MiB combined limit.`, + ) + } + return attachments.map((attachment) => { + const filename = attachment.filename.trim() + if (!filename) throw new Error('Attachment filename is required.') + const contentType = attachment.contentType.trim() + if (!contentType) { + throw new Error(`Attachment content type is required: ${filename}`) + } + const contentBase64 = attachment.contentBase64.trim() + let bytes: Uint8Array + try { + bytes = base64ToBytes(contentBase64) + } catch { + throw new Error(`Attachment content must be valid base64: ${filename}`) + } + if (bytes.byteLength === 0) { + throw new Error(`Attachment content is empty: ${filename}`) + } + return { filename, contentType, bytes, contentBase64 } + }) +} + +/** + * Persist outbound attachment bytes to R2 and record one email_attachments + * row per file. A failed blob put degrades that attachment to metadata-only + * (storage_kind 'unavailable') rather than blocking the send — the provider + * still receives the in-memory bytes. + */ +async function storeOutboundAttachments(input: { + db: D1Database + blobs: R2Bucket + userId: string + messageId: string + attachments: Array +}) { + if (input.attachments.length === 0) return + const rows: Parameters[0]['attachments'] = [] + for (const attachment of input.attachments) { + const attachmentId = crypto.randomUUID() + const storageKey = emailAttachmentBlobKey( + input.userId, + input.messageId, + attachmentId, + ) + let stored = false + try { + await input.blobs.put(storageKey, attachment.bytes, { + httpMetadata: { contentType: attachment.contentType }, + }) + stored = true + } catch (error) { + console.warn( + 'email-outbound-attachment-blob-put-failed', + storageKey, + error, + ) + } + rows.push({ + id: attachmentId, + filename: attachment.filename, + contentType: attachment.contentType, + disposition: 'attachment', + size: attachment.bytes.byteLength, + storageKind: stored ? 'external' : 'unavailable', + storageKey: stored ? storageKey : null, + }) + } + try { + await insertEmailAttachments({ + db: input.db, + messageId: input.messageId, + attachments: rows, + }) + } catch (error) { + // Without rows, cleanup paths (retention, deletion) can never find + // these blobs again — delete them now rather than orphan them. + await Promise.all( + rows + .filter((row) => row.storageKey != null) + .map((row) => + input.blobs.delete(row.storageKey!).catch((deleteError: unknown) => { + console.warn( + 'email-outbound-attachment-blob-cleanup-failed', + row.storageKey, + deleteError, + ) + }), + ), + ) + throw error + } +} + async function requireStoredEmailMessage(input: { env: SendEmailEnv userId: string @@ -170,6 +317,7 @@ async function sendViaBinding(input: { html?: string | null replyTo?: string | null headers: Record + attachments: Array }) { const binding = input.env.EMAIL if (!binding) return { sent: false, messageId: null } @@ -181,6 +329,18 @@ async function sendViaBinding(input: { headers: input.headers, ...(input.text ? { text: input.text } : {}), ...(input.html ? { html: input.html } : {}), + ...(input.attachments.length > 0 + ? { + // The binding treats string content as raw text, so binary + // payloads must go through as bytes rather than base64. + attachments: input.attachments.map((attachment) => ({ + disposition: 'attachment' as const, + filename: attachment.filename, + type: attachment.contentType, + content: attachment.bytes, + })), + } + : {}), }) return { sent: true, messageId: result.messageId ?? null } } @@ -194,6 +354,7 @@ async function sendViaRestFallback(input: { html?: string | null replyTo?: string | null headers: Record + attachments: Array }) { const html = input.html ?? input.text if (!html) { @@ -214,6 +375,19 @@ async function sendViaRestFallback(input: { replyTo: input.replyTo ?? undefined, headers: Object.keys(input.headers).length > 0 ? input.headers : undefined, + attachments: + input.attachments.length > 0 + ? // The REST API expects base64 string content. + input.attachments.map((attachment) => ({ + content: attachment.contentBase64, + filename: attachment.filename, + type: attachment.contentType, + disposition: 'attachment' as const, + // The inferred schema output type requires this key even + // when undefined; JSON.stringify drops it from the payload. + contentId: undefined, + })) + : undefined, }, ) if (!result.ok) { @@ -225,9 +399,10 @@ async function sendViaRestFallback(input: { function outboundEmailContentBytes( text: string | null, html: string | null, + attachmentBytes = 0, ): number | undefined { - if (!text && !html) return undefined - let bytes = 0 + if (!text && !html && attachmentBytes === 0) return undefined + let bytes = attachmentBytes if (text) bytes += new TextEncoder().encode(text).byteLength if (html) bytes += new TextEncoder().encode(html).byteLength return bytes @@ -305,6 +480,26 @@ export async function sendOutboundEmail( const text = input.text?.trim() || null const html = input.html?.trim() || null if (!text && !html) throw new Error('Email text or HTML body is required.') + const attachments = prepareOutboundAttachments(input.attachments) + const attachmentBytesTotal = attachments.reduce( + (total, attachment) => total + attachment.bytes.byteLength, + 0, + ) + if (attachments.length > 0) { + // Attachments put outbound mail under the same per-message ceiling + // as inbound mail. Body-only sends keep their pre-attachment + // behavior (no per-message cap). + await assertWithinEntitlement({ + db: input.env.APP_DB, + userId: input.userId, + email: sender.accountEmail, + resource: 'email_message_bytes', + requested: 0, + getCurrent: async () => + (outboundEmailContentBytes(text, html) ?? 0) + attachmentBytesTotal, + fallbackLimit: nullPlanEmailFallbackLimits.email_message_bytes, + }) + } const messageIdHeader = `<${crypto.randomUUID()}@kody.local>` const storedHeaders = buildStoredHeaders({ messageId: messageIdHeader, @@ -315,18 +510,19 @@ export async function sendOutboundEmail( db: input.env.APP_DB, userId: input.userId, email: sender.accountEmail, - requested: estimateEntitlementStorageEntryBytes({ - key: messageIdHeader, - value: { - from, - to, - subject, - replyTo: input.replyTo, - text, - html, - headers: storedHeaders, - }, - }), + requested: + estimateEntitlementStorageEntryBytes({ + key: messageIdHeader, + value: { + from, + to, + subject, + replyTo: input.replyTo, + text, + html, + headers: storedHeaders, + }, + }) + attachmentBytesTotal, }) // Atomic check-and-increment: the counter tracks attempts for every user @@ -390,6 +586,43 @@ export async function sendOutboundEmail( sentAt: null, }, }) + try { + await storeOutboundAttachments({ + db: input.env.APP_DB, + blobs: input.env.EMAIL_BLOBS, + userId: input.userId, + messageId: message.id, + attachments, + }) + } catch (error) { + // The daily send counter deliberately tracks attempts (see the + // consumeDailyEntitlement comment), but the message must not stay in + // 'stored' limbo with no delivery attempt recorded. + const messageText = getErrorMessage(error) + await updateEmailMessageDelivery({ + db: input.env.APP_DB, + messageId: message.id, + status: 'failed', + providerMessageId: null, + error: messageText, + sentAt: null, + }).catch((updateError) => { + console.warn('email-attachment-failure-status-update-failed', updateError) + }) + await insertEmailDeliveryEvent({ + db: input.env.APP_DB, + messageId: message.id, + userId: input.userId, + inboxId: null, + eventType: 'failed', + provider: 'cloudflare-email', + providerMessageId: null, + detail: { error: messageText }, + }).catch((eventError) => { + console.warn('email-attachment-failure-event-insert-failed', eventError) + }) + throw error + } await insertEmailDeliveryEvent({ db: input.env.APP_DB, messageId: message.id, @@ -398,10 +631,14 @@ export async function sendOutboundEmail( eventType: 'send_requested', provider: 'cloudflare-email', providerMessageId: null, - detail: { to, from, subject }, + detail: { to, from, subject, attachmentCount: attachments.length }, }) - const messageContentBytes = outboundEmailContentBytes(text, html) + const messageContentBytes = outboundEmailContentBytes( + text, + html, + attachmentBytesTotal, + ) const sendStartedAtMs = Date.now() let sendOutcome: 'success' | 'error' = 'success' try { @@ -416,6 +653,7 @@ export async function sendOutboundEmail( ? (normalizeEmailAddress(input.replyTo) ?? undefined) : undefined, headers: providerHeaders, + attachments, }) const providerMessageId = bindingResult.sent ? bindingResult.messageId @@ -430,6 +668,7 @@ export async function sendOutboundEmail( ? (normalizeEmailAddress(input.replyTo) ?? undefined) : undefined, headers: providerHeaders, + attachments, }) await updateEmailMessageDelivery({ db: input.env.APP_DB, diff --git a/packages/worker/src/email/outbound.workers.test.ts b/packages/worker/src/email/outbound.workers.test.ts index 6f84046fa5..f30cdc0e11 100644 --- a/packages/worker/src/email/outbound.workers.test.ts +++ b/packages/worker/src/email/outbound.workers.test.ts @@ -1,14 +1,21 @@ import { env } from 'cloudflare:workers' import { expect, test } from 'vitest' import { http, HttpResponse } from 'msw' +import { bytesToBase64 } from '@kody-internal/shared/base64.ts' import { ensureUsageRollupsTestSchema } from '#worker/usage/test-schema.ts' import { ensureEmailTestSchema } from './test-schema.ts' import { + deleteEmailMessageById, + getEmailAttachmentById, getEmailMessageById, insertEmailMessage, + listEmailAttachmentsForMessage, listEmailMessages, } from './repo.ts' -import { sendOutboundEmail } from './outbound.ts' +import { + maxOutboundEmailAttachmentTotalBytes, + sendOutboundEmail, +} from './outbound.ts' import { createMswNodeServer } from '#worker/test-support/msw-node-server.ts' import { isEntitlementLimitError } from '#worker/entitlements/errors.ts' import { @@ -511,6 +518,264 @@ test('sendOutboundEmail preserves reply headers and records failed fallback send ).toBeUndefined() }) +test('sendOutboundEmail sends, stores, and re-serves reply attachments', async () => { + await ensureEmailTestSchema(env.APP_DB) + await ensureUsageRollupsTestSchema(env.APP_DB) + const accountEmail = `account-${crypto.randomUUID()}@example.com` + const userId = await createStableUserIdFromEmail(accountEmail) + await seedVerifiedAccount({ email: accountEmail }) + const inbound = await insertEmailMessage({ + db: env.APP_DB, + blobs: env.EMAIL_BLOBS, + message: { + direction: 'inbound', + userId, + fromAddress: 'recipient@example.com', + envelopeFrom: 'recipient@example.com', + toAddresses: [accountEmail], + subject: 'Needs a file', + messageIdHeader: '', + processingStatus: 'stored', + receivedAt: new Date().toISOString(), + }, + }) + const pdfBytes = new Uint8Array([0x25, 0x50, 0x44, 0x46, 0x2d, 0x31]) + const sent: Array> = [] + const sendEnv = { + ...env, + APP_BASE_URL: platformBaseUrl, + EMAIL: { + async send(message: Record) { + sent.push(message) + return { messageId: 'provider-attach-1' } + }, + } as unknown as SendEmail, + } + + const result = await sendOutboundEmail({ + env: sendEnv, + userId, + accountEmail, + recipientPolicy: 'reply', + replyToMessageId: inbound.id, + subject: 'Re: Needs a file', + text: 'File attached.', + attachments: [ + { + filename: 'invoice.pdf', + contentType: 'application/pdf', + contentBase64: bytesToBase64(pdfBytes), + }, + ], + }) + + expect(result.status).toBe('sent') + // The binding receives raw bytes (string content would be treated as + // raw text, not base64). + expect(sent).toHaveLength(1) + const sentAttachments = sent[0]?.attachments as Array> + expect(sentAttachments).toHaveLength(1) + expect(sentAttachments[0]).toMatchObject({ + disposition: 'attachment', + filename: 'invoice.pdf', + type: 'application/pdf', + }) + expect(new Uint8Array(sentAttachments[0]?.content as Uint8Array)).toEqual( + pdfBytes, + ) + + // The attachment is stored as its own R2 object and stays readable via + // the normal attachment read path. + const stored = await listEmailAttachmentsForMessage({ + db: env.APP_DB, + messageId: result.message.id, + }) + expect(stored).toHaveLength(1) + expect(stored[0]).toMatchObject({ + filename: 'invoice.pdf', + contentType: 'application/pdf', + disposition: 'attachment', + size: pdfBytes.byteLength, + storageKind: 'external', + }) + const storageKey = stored[0]?.storageKey + expect(storageKey).toContain(`email-attachment:v1:${userId}/`) + const loaded = await getEmailAttachmentById({ + db: env.APP_DB, + blobs: env.EMAIL_BLOBS, + userId, + attachmentId: stored[0]!.id, + }) + expect(loaded?.contentBase64).toBe(bytesToBase64(pdfBytes)) + + // Usage bytes include the attachment payload. + const rollups = await listEmailSendRollups(env.APP_DB, userId) + expect(rollups[0]).toMatchObject({ + total_bytes: + new TextEncoder().encode('File attached.').byteLength + + pdfBytes.byteLength, + }) + + // Deleting the message removes the external attachment blob too. + await deleteEmailMessageById({ + db: env.APP_DB, + blobs: env.EMAIL_BLOBS, + messageId: result.message.id, + }) + expect(await env.EMAIL_BLOBS.get(storageKey!)).toBeNull() + expect( + await listEmailAttachmentsForMessage({ + db: env.APP_DB, + messageId: result.message.id, + }), + ).toEqual([]) +}) + +test('sendOutboundEmail passes base64 attachments to the REST fallback', async () => { + await ensureEmailTestSchema(env.APP_DB) + const accountEmail = `account-${crypto.randomUUID()}@example.com` + const userId = await createStableUserIdFromEmail(accountEmail) + await seedVerifiedAccount({ email: accountEmail }) + const contentBase64 = bytesToBase64(new TextEncoder().encode('name,total')) + const fetchCalls: Array> = [] + + using _server = createMswNodeServer( + [ + http.post(cloudflareEmailApi, async ({ request }) => { + fetchCalls.push((await request.json()) as Record) + return HttpResponse.json({ + success: true, + result: { message_id: 'rest-attach-1' }, + }) + }), + ], + { onUnhandledRequest: 'bypass' }, + ) + + const result = await sendOutboundEmail({ + env: { + ...env, + APP_BASE_URL: platformBaseUrl, + EMAIL: undefined as unknown as SendEmail, + CLOUDFLARE_ACCOUNT_ID: 'account-123', + CLOUDFLARE_API_BASE_URL: 'https://api.cloudflare.test', + CLOUDFLARE_API_TOKEN: 'token-123', + }, + userId, + accountEmail, + recipientPolicy: 'self', + subject: 'Report', + text: 'Report attached.', + attachments: [ + { + filename: 'report.csv', + contentType: 'text/csv', + contentBase64, + }, + ], + }) + + expect(result.status).toBe('sent') + expect(result.providerMessageId).toBe('rest-attach-1') + expect(fetchCalls).toHaveLength(1) + expect(fetchCalls[0]?.attachments).toEqual([ + { + content: contentBase64, + filename: 'report.csv', + type: 'text/csv', + disposition: 'attachment', + }, + ]) +}) + +test('sendOutboundEmail rejects invalid and oversized attachments', async () => { + await ensureEmailTestSchema(env.APP_DB) + const accountEmail = `account-${crypto.randomUUID()}@example.com` + const userId = await createStableUserIdFromEmail(accountEmail) + await seedVerifiedAccount({ email: accountEmail }) + + await expect( + sendOutboundEmail({ + env: createBindingSendEnv(), + userId, + accountEmail, + recipientPolicy: 'self', + subject: 'Bad base64', + text: 'Body', + attachments: [ + { + filename: 'broken.bin', + contentType: 'application/octet-stream', + contentBase64: '!!not-base64!!', + }, + ], + }), + ).rejects.toThrow('Attachment content must be valid base64: broken.bin') + + // Attachments put the message under the plan-less per-message backstop + // (email_message_bytes) that body-only sends are not subject to. + const oversize = new Uint8Array( + nullPlanEmailFallbackLimits.email_message_bytes + 1, + ) + const error = await sendOutboundEmail({ + env: createBindingSendEnv(), + userId, + accountEmail, + recipientPolicy: 'self', + subject: 'Too big', + text: 'Body', + attachments: [ + { + filename: 'huge.bin', + contentType: 'application/octet-stream', + contentBase64: bytesToBase64(oversize), + }, + ], + }).then( + () => null, + (thrown: unknown) => thrown, + ) + if (!isEntitlementLimitError(error)) { + throw new Error('Expected an EntitlementLimitError from sendOutboundEmail.') + } + expect(error.details).toMatchObject({ + resource: 'email_message_bytes', + }) + // Attachments whose base64 length alone exceeds the 5 MiB hard cap are + // rejected before any decoding (no entitlement lookup, no allocation). + const hardCapBase64Length = Math.ceil( + ((maxOutboundEmailAttachmentTotalBytes + 3) * 4) / 3, + ) + await expect( + sendOutboundEmail({ + env: createBindingSendEnv(), + userId, + accountEmail, + recipientPolicy: 'self', + subject: 'Way too big', + text: 'Body', + attachments: [ + { + filename: 'giant.bin', + contentType: 'application/octet-stream', + contentBase64: 'A'.repeat(hardCapBase64Length), + }, + ], + }), + ).rejects.toThrow('exceed the 5 MiB combined limit') + + // None of the failures consumed the daily send quota or stored a message. + expect(await readDailyEmailSendCounter(userId)).toBe(0) + expect( + await listEmailMessages({ + db: env.APP_DB, + userId, + direction: 'outbound', + limit: 5, + }), + ).toEqual([]) +}) + test('sendOutboundEmail enforces email_sends_per_day for plan users at the limit', async () => { await ensureEmailTestSchema(env.APP_DB) const email = `planned-${crypto.randomUUID()}@example.com` diff --git a/packages/worker/src/email/repo.ts b/packages/worker/src/email/repo.ts index 6bec6f5ca8..1880a7f89c 100644 --- a/packages/worker/src/email/repo.ts +++ b/packages/worker/src/email/repo.ts @@ -27,6 +27,20 @@ export function emailRawMimeKey(userId: string, messageId: string) { return `email-raw:v1:${userId}/${messageId}` } +/** + * R2 object key for an attachment stored on its own (storage_kind + * 'external'), used by outbound mail whose bytes never exist as raw MIME. + * The userId prefix follows the same per-user isolation contract as + * emailRawMimeKey. + */ +export function emailAttachmentBlobKey( + userId: string, + messageId: string, + attachmentId: string, +) { + return `email-attachment:v1:${userId}/${messageId}/${attachmentId}` +} + /** * Best-effort raw-MIME offload to R2. Returns the stored object key, or * null when the payload must stay inline in D1 because the put failed @@ -886,33 +900,61 @@ export async function searchEmailMessages(input: { export async function deleteEmailMessageById(input: { db: D1Database - /** EMAIL_BLOBS bucket; the raw-MIME blob is deleted with the row. */ + /** + * EMAIL_BLOBS bucket; the raw-MIME blob and any external attachment + * blobs are deleted with the rows. + */ blobs: R2Bucket messageId: string }) { - // Capture the blob key before the row disappears. A failed read must - // abort the delete (and be retried) rather than orphan the R2 blob; only - // the R2 delete itself is best-effort. + // Capture the blob keys before the rows disappear. A failed read must + // abort the delete (and be retried) rather than orphan the R2 blobs; + // only the R2 deletes themselves are best-effort. const row = await input.db .prepare(`SELECT raw_mime_key FROM email_messages WHERE id = ?`) .bind(input.messageId) .first<{ raw_mime_key: string | null }>() const rawMimeKey = row?.raw_mime_key ?? null - await input.db - .prepare(`DELETE FROM email_messages WHERE id = ?`) + const attachmentRows = await input.db + .prepare( + `SELECT storage_key FROM email_attachments + WHERE message_id = ? AND storage_key IS NOT NULL`, + ) .bind(input.messageId) - .run() - if (rawMimeKey != null) { - await input.blobs.delete(rawMimeKey).catch((error: unknown) => { - console.warn('email-raw-mime-blob-delete-failed', rawMimeKey, error) + .all<{ storage_key: string }>() + const blobKeys = [ + ...(rawMimeKey != null ? [rawMimeKey] : []), + ...(attachmentRows.results ?? []).map((result) => result.storage_key), + ] + // Atomic batch: a partial delete (attachments gone, message left) would + // lose the storage_key values needed to delete the R2 blobs on retry. + await input.db.batch([ + input.db + .prepare(`DELETE FROM email_attachments WHERE message_id = ?`) + .bind(input.messageId), + input.db + .prepare(`DELETE FROM email_messages WHERE id = ?`) + .bind(input.messageId), + ]) + for (const blobKey of blobKeys) { + await input.blobs.delete(blobKey).catch((error: unknown) => { + console.warn('email-blob-delete-failed', blobKey, error) }) } } +/** Rows per D1 batch when inserting attachment metadata. */ +const emailAttachmentInsertBatchSize = 50 + export async function insertEmailAttachments(input: { db: D1Database messageId: string attachments: Array<{ + /** + * Pre-generated row id. Callers that store attachment bytes in R2 + * before inserting the row pass the id embedded in the blob key. + */ + id?: string | null filename?: string | null contentType?: string | null contentId?: string | null @@ -922,28 +964,40 @@ export async function insertEmailAttachments(input: { storageKey?: string | null }> }) { + if (input.attachments.length === 0) return const timestamp = nowIso() - for (const attachment of input.attachments) { - await input.db - .prepare( - `INSERT INTO email_attachments ( - id, message_id, filename, content_type, content_id, disposition, - size, storage_kind, storage_key, created_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - ) - .bind( - crypto.randomUUID(), - input.messageId, - attachment.filename ?? null, - attachment.contentType ?? null, - attachment.contentId ?? null, - attachment.disposition ?? null, - attachment.size ?? null, - attachment.storageKind, - attachment.storageKey ?? null, - timestamp, - ) - .run() + // Batched inserts keep typical attachment sets (outbound caps at ten + // files) all-or-nothing, while chunking protects large inbound MIME + // from oversized D1 batches. On a mid-chunk failure, callers that need + // consistency delete by message_id, which covers earlier chunks too. + const statement = input.db.prepare( + `INSERT INTO email_attachments ( + id, message_id, filename, content_type, content_id, disposition, + size, storage_kind, storage_key, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + const statements = input.attachments.map((attachment) => + statement.bind( + attachment.id ?? crypto.randomUUID(), + input.messageId, + attachment.filename ?? null, + attachment.contentType ?? null, + attachment.contentId ?? null, + attachment.disposition ?? null, + attachment.size ?? null, + attachment.storageKind, + attachment.storageKey ?? null, + timestamp, + ), + ) + for ( + let index = 0; + index < statements.length; + index += emailAttachmentInsertBatchSize + ) { + await input.db.batch( + statements.slice(index, index + emailAttachmentInsertBatchSize), + ) } } export async function insertEmailMessageWithAttachments( @@ -1005,6 +1059,25 @@ export async function getEmailAttachmentById(input: { .first>() if (!row) return null const attachment = mapAttachmentRow(row) + // Externally stored attachments (outbound mail) keep their bytes in a + // dedicated R2 object instead of inside raw MIME. + if (attachment.storageKind === 'external') { + const object = attachment.storageKey + ? await input.blobs.get(attachment.storageKey) + : null + if (!object) { + return { ...attachment, content: null, contentBase64: null } + } + const buffer = await object.arrayBuffer() + return { + ...attachment, + content: buffer, + contentBase64: bytesToBase64(new Uint8Array(buffer)), + } + } + if (attachment.storageKind === 'unavailable') { + return { ...attachment, content: null, contentBase64: null } + } const message = await getEmailMessageById({ db: input.db, userId: input.userId, diff --git a/packages/worker/src/mcp/capabilities/email/email-reply.node.test.ts b/packages/worker/src/mcp/capabilities/email/email-reply.node.test.ts index bc4fc55d12..c01f814b04 100644 --- a/packages/worker/src/mcp/capabilities/email/email-reply.node.test.ts +++ b/packages/worker/src/mcp/capabilities/email/email-reply.node.test.ts @@ -11,6 +11,7 @@ vi.mock('#worker/email/repo.ts', () => ({ })) vi.mock('#worker/email/outbound.ts', () => ({ + maxOutboundEmailAttachments: 10, sendOutboundEmail: mocks.sendOutboundEmail, })) @@ -128,3 +129,68 @@ test('email_reply throws when provider delivery is persisted as failed', async ( expect.objectContaining({ to: expect.anything() }), ) }) + +test('email_reply forwards attachments to the outbound send', async () => { + mocks.getEmailMessageById.mockResolvedValue({ + id: 'inbound-1', + subject: 'Hello', + fromAddress: 'sender@example.com', + envelopeFrom: null, + replyToAddresses: [], + messageIdHeader: '', + references: [], + threadId: 'thread-1', + inboxId: 'inbox-1', + }) + mocks.sendOutboundEmail.mockResolvedValue({ + status: 'sent', + error: null, + providerMessageId: 'provider-1', + message: { + id: 'outbound-1', + direction: 'outbound', + inboxId: 'inbox-1', + threadId: 'thread-1', + fromAddress: 'user@heykody.dev', + envelopeFrom: 'user@heykody.dev', + toAddresses: ['sender@example.com'], + subject: 'Re: Hello', + messageIdHeader: '', + processingStatus: 'sent', + providerMessageId: 'provider-1', + error: null, + receivedAt: null, + sentAt: '2026-05-13T07:30:16.000Z', + createdAt: '2026-05-13T07:30:16.000Z', + updatedAt: '2026-05-13T07:30:16.000Z', + }, + }) + + await emailReplyCapability.handler( + { + message_id: 'inbound-1', + text: 'Reply body', + attachments: [ + { + filename: 'notes.txt', + content_type: 'text/plain', + content_base64: 'aGVsbG8=', + }, + ], + }, + createContext(), + ) + expect(mocks.sendOutboundEmail).toHaveBeenCalledWith( + expect.objectContaining({ + recipientPolicy: 'reply', + replyToMessageId: 'inbound-1', + attachments: [ + { + filename: 'notes.txt', + contentType: 'text/plain', + contentBase64: 'aGVsbG8=', + }, + ], + }), + ) +}) diff --git a/packages/worker/src/mcp/capabilities/email/email-reply.ts b/packages/worker/src/mcp/capabilities/email/email-reply.ts index e8a554a4a9..953e1b6999 100644 --- a/packages/worker/src/mcp/capabilities/email/email-reply.ts +++ b/packages/worker/src/mcp/capabilities/email/email-reply.ts @@ -2,7 +2,10 @@ import { z } from 'zod' import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' import { requireVerifiedEmailAccountUser } from './require-verified-user.ts' -import { sendOutboundEmail } from '#worker/email/outbound.ts' +import { + maxOutboundEmailAttachments, + sendOutboundEmail, +} from '#worker/email/outbound.ts' import { getEmailMessageById } from '#worker/email/repo.ts' import { emailMessageSummarySchema, @@ -15,8 +18,8 @@ export const emailReplyCapability = defineDomainCapability( { name: 'email_reply', description: - 'Reply to a stored inbound email from your platform-assigned {username}@ address, preserving thread headers. The recipient always comes from the stored message.', - keywords: ['email', 'reply', 'thread', 'message'], + 'Reply to a stored inbound email from your platform-assigned {username}@ address, preserving thread headers and optionally attaching files (base64 content). The recipient always comes from the stored message.', + keywords: ['email', 'reply', 'thread', 'message', 'attachment'], readOnly: false, idempotent: false, destructive: false, @@ -25,6 +28,17 @@ export const emailReplyCapability = defineDomainCapability( message_id: z.string().min(1), text: z.string().min(1).optional(), html: z.string().min(1).optional(), + attachments: z + .array( + z.object({ + filename: z.string().min(1).max(255), + content_type: z.string().min(1).max(255), + content_base64: z.string().min(1), + }), + ) + .min(1) + .max(maxOutboundEmailAttachments) + .optional(), }) .refine((value) => value.text !== undefined || value.html !== undefined, { message: 'Email text or HTML body is required.', @@ -53,6 +67,11 @@ export const emailReplyCapability = defineDomainCapability( : `Re: ${original.subject ?? '(no subject)'}`, text: args.text ?? null, html: args.html ?? null, + attachments: args.attachments?.map((attachment) => ({ + filename: attachment.filename, + contentType: attachment.content_type, + contentBase64: attachment.content_base64, + })), inReplyToHeader: original.messageIdHeader ?? null, references: [ ...stringArray(original.references),