diff --git a/docs/contributing/setup-manifest.md b/docs/contributing/setup-manifest.md index 1a01d5019e..faad3a159e 100644 --- a/docs/contributing/setup-manifest.md +++ b/docs/contributing/setup-manifest.md @@ -11,6 +11,16 @@ This project uses the following resources: - KV namespace for OAuth/session storage - `binding`: `OAUTH_KV` - `title`: `-oauth` +- Cloudflare Email Sending / Email Service Worker binding + - `binding`: `EMAIL` + - `wrangler` key: `send_email` + - Production domains still need Cloudflare-side sender/domain verification + before sends succeed. +- Cloudflare Email Routing for inbound mail + - Configure MX records and selected route aliases in the Cloudflare + dashboard. + - Route only test/storage aliases to the Worker until quarantine behavior is + verified. - Vectorize indexes for MCP capability search (`CAPABILITY_VECTOR_INDEX`) - Production: `kody-capabilities-prod` - Preview: `kody-capabilities-preview` @@ -67,6 +77,9 @@ automatically: - `APP_COMMIT_SHA` (optional; set automatically by deploy workflows for version-aware `/health` checks) - `CLOUDFLARE_EMAIL_FROM` (optional; sender address for outbound email) + Existing app email uses this as a default sender. First-class MCP + `email_send` additionally requires a verified sender identity created through + the email capabilities. - `AI_GATEWAY_ID` (required when `AI_MODE=remote`; deploy workflows sync a gateway ID from GitHub Actions secrets so remote inference goes through Cloudflare AI Gateway) @@ -114,6 +127,8 @@ Configure these GitHub Actions secrets and variables for workflows: - `AI_GATEWAY_ID_PREVIEW` (required for preview deploys that use remote AI inference) - `CLOUDFLARE_EMAIL_FROM` (optional, required to send app email) + First-class Kody email also requires the `EMAIL` send binding plus + Cloudflare Email Service sender/domain verification. - `SENTRY_DSN` (optional; create a JavaScript/Cloudflare project in Sentry and paste the DSN; syncs to the Worker as a secret when set in GitHub Actions) - `CAPABILITY_REINDEX_SECRET` (optional; triggers post-deploy Vectorize reindex diff --git a/docs/use/email-primitives.md b/docs/use/email-primitives.md new file mode 100644 index 0000000000..6392b419e1 --- /dev/null +++ b/docs/use/email-primitives.md @@ -0,0 +1,51 @@ +# Email primitives + +Kody has a storage-only email surface for Cloudflare Email Service and Email +Routing. It can send from verified identities, receive routed mail, store parsed +messages, and quarantine unknown senders. + +## Capabilities + +Use the MCP `email` domain: + +- `email_inbox_create` creates an inbox and routable alias. +- `email_inbox_list` lists inboxes and aliases for the signed-in user. +- `email_sender_approve` verifies an outbound sender identity or allowlists an + inbound sender/domain for an inbox. +- `email_sender_revoke` disables an allow rule. +- `email_policy_get` inspects effective sender policy. +- `email_send` sends outbound mail from a verified sender identity. +- `email_reply` replies to a stored inbound message. +- `email_message_list` lists stored sent, accepted, quarantined, or failed + messages. +- `email_message_get` returns parsed bodies, headers, thread metadata, and + attachment metadata. + +## Safety model + +- Unknown inbound senders are quarantined. +- Display names are not trusted. Kody stores envelope sender, parsed `From`, and + authentication headers separately. +- Outbound sending requires a verified sender identity. +- Inbound package handlers are intentionally disabled in this first slice. Mail + is stored and audited only. +- Attachments are metadata-only for now; raw MIME for small messages is stored so + the first pass can be tested locally. + +## Local inbound testing + +Run the worker locally, create an inbox alias, then post raw MIME to Wrangler's +email test endpoint: + +```sh +curl --request POST \ + 'http://localhost:8787/cdn-cgi/handler/email?from=sender@example.com&to=alias@example.com' \ + --data-raw 'From: Sender +To: Alias +Subject: Hello +Message-ID: + +Hello from local email routing.' +``` + +Then inspect the message with `email_message_list` and `email_message_get`. diff --git a/package-lock.json b/package-lock.json index d4e7766372..255ab80694 100644 --- a/package-lock.json +++ b/package-lock.json @@ -17094,6 +17094,12 @@ "node": ">= 0.4" } }, + "node_modules/postal-mime": { + "version": "2.7.4", + "resolved": "https://registry.npmjs.org/postal-mime/-/postal-mime-2.7.4.tgz", + "integrity": "sha512-0WdnFQYUrPGGTFu1uOqD2s7omwua8xaeYGdO6rb88oD5yJ/4pPHDA4sdWqfD8wQVfCny563n/HQS7zTFft+f/g==", + "license": "MIT-0" + }, "node_modules/postcss": { "version": "8.5.8", "devOptional": true, @@ -21185,6 +21191,7 @@ "acorn": "^8.16.0", "agents": "^0.11.1", "croner": "^10.0.1", + "postal-mime": "^2.7.4", "remix": "3.0.0-alpha.3", "workers-ai-provider": "^3.1.11", "zod": "^4.3.6" diff --git a/packages/mock-servers/cloudflare/src/worker.ts b/packages/mock-servers/cloudflare/src/worker.ts index 088f1047e3..d21da3be04 100644 --- a/packages/mock-servers/cloudflare/src/worker.ts +++ b/packages/mock-servers/cloudflare/src/worker.ts @@ -589,6 +589,8 @@ async function handleEmailSend( return envelope( { delivered: Array.isArray(payload.to) ? payload.to : [payload.to], + message_id: messageId, + messageId, permanent_bounces: [], queued: [], }, diff --git a/packages/shared/src/outbound-email.ts b/packages/shared/src/outbound-email.ts index 5eb66acda8..f616a98e1e 100644 --- a/packages/shared/src/outbound-email.ts +++ b/packages/shared/src/outbound-email.ts @@ -1,5 +1,7 @@ import { array, + createSchema, + fail, object, optional, string, @@ -10,12 +12,38 @@ import { minLength } from 'remix/data-schema/checks' const nonEmptyStringSchema = string().pipe(minLength(1)) +const optionalHeadersSchema = createSchema< + unknown, + Record | undefined +>((value, context) => { + if (value === undefined) return { value: undefined } + if ( + !value || + typeof value !== 'object' || + Array.isArray(value) || + (Object.getPrototypeOf(value) !== Object.prototype && + Object.getPrototypeOf(value) !== null) + ) { + return fail('Expected headers object', context.path) + } + const headers: Record = {} + for (const [key, headerValue] of Object.entries(value)) { + if (typeof headerValue !== 'string') { + return fail(`Expected string header value for "${key}"`, context.path) + } + headers[key] = headerValue + } + return { value: headers } +}) + const outboundEmailSchema = object({ from: nonEmptyStringSchema, to: union([nonEmptyStringSchema, array(nonEmptyStringSchema)]), subject: nonEmptyStringSchema, html: nonEmptyStringSchema, text: optional(nonEmptyStringSchema), + replyTo: optional(nonEmptyStringSchema), + headers: optionalHeadersSchema, }) export type OutboundEmail = InferOutput diff --git a/packages/worker/migrations/0030-email-primitives.sql b/packages/worker/migrations/0030-email-primitives.sql new file mode 100644 index 0000000000..cfbe51e4e7 --- /dev/null +++ b/packages/worker/migrations/0030-email-primitives.sql @@ -0,0 +1,200 @@ +CREATE TABLE IF NOT EXISTS email_sender_identities ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + package_id TEXT, + email TEXT NOT NULL, + domain TEXT, + display_name TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL CHECK (status IN ('pending', 'verified', 'disabled')), + verified_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_email_sender_identities_user_email +ON email_sender_identities(user_id, email); + +CREATE INDEX IF NOT EXISTS idx_email_sender_identities_user_domain +ON email_sender_identities(user_id, domain); + +CREATE TABLE IF NOT EXISTS email_inboxes ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + package_id TEXT, + name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + mode TEXT NOT NULL CHECK (mode IN ('quarantine', 'accept')), + enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_email_inboxes_user_name +ON email_inboxes(user_id, name); + +CREATE INDEX IF NOT EXISTS idx_email_inboxes_user_created_at +ON email_inboxes(user_id, created_at); + +CREATE TABLE IF NOT EXISTS email_inbox_addresses ( + id TEXT PRIMARY KEY, + inbox_id TEXT NOT NULL, + user_id TEXT NOT NULL, + address TEXT NOT NULL, + local_part TEXT NOT NULL, + domain TEXT NOT NULL, + reply_token_hash TEXT, + enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (inbox_id) REFERENCES email_inboxes(id) ON DELETE CASCADE +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_email_inbox_addresses_address +ON email_inbox_addresses(address); + +CREATE INDEX IF NOT EXISTS idx_email_inbox_addresses_inbox_id +ON email_inbox_addresses(inbox_id); + +CREATE INDEX IF NOT EXISTS idx_email_inbox_addresses_reply_token +ON email_inbox_addresses(reply_token_hash); + +CREATE TABLE IF NOT EXISTS email_threads ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + inbox_id TEXT, + subject_normalized TEXT NOT NULL DEFAULT '', + root_message_id_header TEXT, + last_message_at TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_email_threads_user_last_message_at +ON email_threads(user_id, last_message_at); + +CREATE INDEX IF NOT EXISTS idx_email_threads_root_message_id +ON email_threads(root_message_id_header); + +CREATE TABLE IF NOT EXISTS email_messages ( + id TEXT PRIMARY KEY, + direction TEXT NOT NULL CHECK (direction IN ('inbound', 'outbound')), + user_id TEXT NOT NULL, + inbox_id TEXT, + thread_id TEXT, + sender_identity_id TEXT, + from_address TEXT NOT NULL, + envelope_from TEXT, + to_addresses_json TEXT NOT NULL DEFAULT '[]', + cc_addresses_json TEXT NOT NULL DEFAULT '[]', + bcc_addresses_json TEXT NOT NULL DEFAULT '[]', + reply_to_addresses_json TEXT NOT NULL DEFAULT '[]', + subject TEXT NOT NULL DEFAULT '', + message_id_header TEXT, + in_reply_to_header TEXT, + references_json TEXT NOT NULL DEFAULT '[]', + headers_json TEXT NOT NULL DEFAULT '{}', + auth_results TEXT, + text_body TEXT, + html_body TEXT, + raw_mime TEXT, + raw_size INTEGER NOT NULL DEFAULT 0, + policy_decision TEXT NOT NULL CHECK (policy_decision IN ('accepted', 'quarantined', 'rejected')), + processing_status TEXT NOT NULL CHECK (processing_status IN ('stored', 'sent', 'failed', 'rejected')), + provider_message_id TEXT, + error TEXT, + received_at TEXT, + sent_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (inbox_id) REFERENCES email_inboxes(id) ON DELETE SET NULL, + FOREIGN KEY (thread_id) REFERENCES email_threads(id) ON DELETE SET NULL, + FOREIGN KEY (sender_identity_id) REFERENCES email_sender_identities(id) ON DELETE SET NULL +); + +CREATE INDEX IF NOT EXISTS idx_email_messages_user_created_at +ON email_messages(user_id, created_at); + +CREATE INDEX IF NOT EXISTS idx_email_messages_inbox_created_at +ON email_messages(inbox_id, created_at); + +CREATE INDEX IF NOT EXISTS idx_email_messages_thread_created_at +ON email_messages(thread_id, created_at); + +CREATE INDEX IF NOT EXISTS idx_email_messages_policy_created_at +ON email_messages(policy_decision, created_at); + +CREATE INDEX IF NOT EXISTS idx_email_messages_message_id_header +ON email_messages(message_id_header); + +CREATE TABLE IF NOT EXISTS email_attachments ( + id TEXT PRIMARY KEY, + message_id TEXT NOT NULL, + filename TEXT, + content_type TEXT NOT NULL, + content_id TEXT, + disposition TEXT, + size INTEGER NOT NULL DEFAULT 0, + storage_kind TEXT NOT NULL CHECK (storage_kind IN ('raw-mime', 'external', 'unavailable')), + storage_key TEXT, + created_at TEXT NOT NULL, + FOREIGN KEY (message_id) REFERENCES email_messages(id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_email_attachments_message_id +ON email_attachments(message_id); + +CREATE TABLE IF NOT EXISTS email_sender_policies ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + inbox_id TEXT, + package_id TEXT, + kind TEXT NOT NULL CHECK (kind IN ('sender', 'domain', 'reply_token')), + value TEXT NOT NULL, + effect TEXT NOT NULL CHECK (effect IN ('allow', 'quarantine', 'reject')), + enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (inbox_id) REFERENCES email_inboxes(id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_email_sender_policies_inbox_kind_value +ON email_sender_policies(inbox_id, kind, value); + +CREATE INDEX IF NOT EXISTS idx_email_sender_policies_user_kind_value +ON email_sender_policies(user_id, kind, value); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_email_sender_policies_user_kind_value_inbox_null_package_null +ON email_sender_policies(user_id, kind, value) +WHERE inbox_id IS NULL AND package_id IS NULL; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_email_sender_policies_user_kind_value_inbox_package_null +ON email_sender_policies(user_id, kind, value, inbox_id) +WHERE inbox_id IS NOT NULL AND package_id IS NULL; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_email_sender_policies_user_kind_value_package_inbox_null +ON email_sender_policies(user_id, kind, value, package_id) +WHERE inbox_id IS NULL AND package_id IS NOT NULL; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_email_sender_policies_user_kind_value_inbox_package +ON email_sender_policies(user_id, kind, value, inbox_id, package_id) +WHERE inbox_id IS NOT NULL AND package_id IS NOT NULL; + +CREATE TABLE IF NOT EXISTS email_delivery_events ( + id TEXT PRIMARY KEY, + message_id TEXT, + user_id TEXT, + inbox_id TEXT, + event_type TEXT NOT NULL CHECK (event_type IN ('receive_started', 'received', 'quarantined', 'rejected', 'send_requested', 'sent', 'failed', 'policy_matched')), + provider TEXT NOT NULL DEFAULT 'kody', + provider_message_id TEXT, + detail_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + FOREIGN KEY (message_id) REFERENCES email_messages(id) ON DELETE SET NULL, + FOREIGN KEY (inbox_id) REFERENCES email_inboxes(id) ON DELETE SET NULL +); + +CREATE INDEX IF NOT EXISTS idx_email_delivery_events_message_id +ON email_delivery_events(message_id); + +CREATE INDEX IF NOT EXISTS idx_email_delivery_events_user_created_at +ON email_delivery_events(user_id, created_at); diff --git a/packages/worker/package.json b/packages/worker/package.json index a90a696df0..f4e57d79ab 100644 --- a/packages/worker/package.json +++ b/packages/worker/package.json @@ -22,6 +22,7 @@ "acorn": "^8.16.0", "agents": "^0.11.1", "croner": "^10.0.1", + "postal-mime": "^2.7.4", "remix": "3.0.0-alpha.3", "workers-ai-provider": "^3.1.11", "zod": "^4.3.6" diff --git a/packages/worker/src/app/email/cloudflare-email.node.test.ts b/packages/worker/src/app/email/cloudflare-email.node.test.ts index 8b5929233a..b2a45c162a 100644 --- a/packages/worker/src/app/email/cloudflare-email.node.test.ts +++ b/packages/worker/src/app/email/cloudflare-email.node.test.ts @@ -99,9 +99,10 @@ test('sendCloudflareEmail posts to the mock Cloudflare email API', async () => { }, ) - expect(sendResult).toEqual({ + expect(sendResult).toMatchObject({ ok: true, }) + expect(sendResult.messageId).toMatch(/^email_/) const response = await fetch(`${mock.origin}/__mocks/messages?token=${token}`) expect(response.status).toBe(200) @@ -156,7 +157,7 @@ test('sendCloudflareEmail defaults the API base URL when it is unset', async () }, ) - expect(result).toEqual({ + expect(result).toMatchObject({ ok: true, }) expect(fetchSpy).toHaveBeenCalledTimes(1) diff --git a/packages/worker/src/app/email/cloudflare-email.ts b/packages/worker/src/app/email/cloudflare-email.ts index a8a405b5be..958163b084 100644 --- a/packages/worker/src/app/email/cloudflare-email.ts +++ b/packages/worker/src/app/email/cloudflare-email.ts @@ -19,6 +19,7 @@ type CloudflareApiEnvelope = { message?: string }> result?: { + message_id?: string delivered?: string[] permanent_bounces?: string[] queued?: string[] @@ -29,6 +30,7 @@ type CloudflareSendResult = { ok: boolean skipped?: boolean error?: string + messageId?: string | null } function normalizeEmailPayload(message: OutboundEmail) { @@ -49,6 +51,12 @@ function normalizeEmailPayload(message: OutboundEmail) { } function logSkippedEmail(reason: string, message: OutboundEmail) { + const headers = + message.headers ? + Object.fromEntries( + Object.keys(message.headers).map((key) => [key, '[REDACTED]']), + ) + : undefined console.warn( reason, JSON.stringify({ @@ -57,6 +65,8 @@ function logSkippedEmail(reason: string, message: OutboundEmail) { subject: message.subject, html: message.html, text: message.text, + replyTo: message.replyTo, + headers, }), ) } @@ -120,14 +130,20 @@ async function sendViaCloudflareApi( return { ok: true, + messageId: payload.result?.message_id ?? null, } } export async function sendCloudflareEmail( config: CloudflareEmailClientConfig, - message: OutboundEmail, + message: Omit & + Partial>, ): Promise { - const normalized = normalizeEmailPayload(message) + const normalized = normalizeEmailPayload({ + ...message, + replyTo: message.replyTo, + headers: message.headers, + }) const apiBaseUrl = typeof config.apiBaseUrl === 'string' && config.apiBaseUrl.trim().length > 0 ? config.apiBaseUrl.trim() diff --git a/packages/worker/src/email/address.node.test.ts b/packages/worker/src/email/address.node.test.ts new file mode 100644 index 0000000000..e3f249a3ae --- /dev/null +++ b/packages/worker/src/email/address.node.test.ts @@ -0,0 +1,37 @@ +import { expect, test } from 'vitest' +import { + extractReplyToken, + findReplyTokenHash, + getEmailDomain, + getEmailLocalPart, + hashReplyToken, + normalizeEmailAddress, + normalizeSubject, + parseHeaderAddressList, +} from './address.ts' + +test('email address helpers normalize mailbox strings and reply tokens', async () => { + expect(normalizeEmailAddress('Alice Example ')).toBe( + 'alice@example.com', + ) + expect(parseHeaderAddressList('A , b@example.net')).toEqual([ + { name: null, address: 'a@example.com' }, + { name: null, address: 'b@example.net' }, + ]) + expect(getEmailLocalPart('Support@Example.com')).toBe('support') + expect(getEmailDomain('Support@Example.com')).toBe('example.com') + expect(normalizeSubject(' Re: Fwd: Hello world ')).toBe('hello world') + + const headers = new Headers({ 'X-Kody-Reply-Token': 'token-123' }) + expect(extractReplyToken({ headers, recipients: [] })).toBe('token-123') + expect(await findReplyTokenHash({ headers, recipients: [] })).toBe( + await hashReplyToken('token-123'), + ) + const alternateHeaders = new Headers({ 'X-Reply-Token': 'alternate-token' }) + expect(extractReplyToken({ headers: alternateHeaders, recipients: [] })).toBe( + 'alternate-token', + ) + expect(await findReplyTokenHash({ headers: alternateHeaders, recipients: [] })).toBe( + await hashReplyToken('alternate-token'), + ) +}) diff --git a/packages/worker/src/email/address.ts b/packages/worker/src/email/address.ts new file mode 100644 index 0000000000..c428f96f74 --- /dev/null +++ b/packages/worker/src/email/address.ts @@ -0,0 +1,117 @@ +import { toHex } from '@kody-internal/shared/hex.ts' + +const mailboxAddressPattern = /?/i + +export function normalizeEmailAddress(value: string) { + const match = value.trim().match(mailboxAddressPattern) + const address = match?.[1]?.trim().toLowerCase() ?? '' + return address.length > 0 ? address : null +} + +export function requireNormalizedEmailAddress(value: string, label = 'Email') { + const normalized = normalizeEmailAddress(value) + if (!normalized) { + throw new Error(`${label} must be a valid email address.`) + } + return normalized +} + +export function getEmailDomain(address: string) { + const normalized = requireNormalizedEmailAddress(address) + const at = normalized.lastIndexOf('@') + return normalized.slice(at + 1) +} + +export function getEmailLocalPart(address: string) { + const normalized = requireNormalizedEmailAddress(address) + const at = normalized.lastIndexOf('@') + return normalized.slice(0, at) +} + +export function normalizeEmailAddressList(values: ReadonlyArray) { + return Array.from( + new Set( + values + .map((value) => normalizeEmailAddress(value)) + .filter((value): value is string => value !== null), + ), + ) +} + +export function parseHeaderEmailAddressList(value: string | null | undefined) { + if (!value) return [] + return normalizeEmailAddressList(value.split(',')) +} + +export function parseHeaderAddressList(value: string | null | undefined) { + return parseHeaderEmailAddressList(value).map((address) => ({ + name: null, + address, + })) +} + +export function normalizeSubject(subject: string | null | undefined) { + const trimmed = subject?.trim() ?? '' + return trimmed + .replace(/^(?:(?:re|fw|fwd):\s*)+/i, '') + .replace(/\s+/g, ' ') + .toLowerCase() +} + +export function createReplyToken() { + const bytes = new Uint8Array(24) + crypto.getRandomValues(bytes) + return toHex(bytes) +} + +export async function hashReplyToken(token: string) { + const digest = await crypto.subtle.digest( + 'SHA-256', + new TextEncoder().encode(token.trim()), + ) + return toHex(new Uint8Array(digest)) +} + +export async function findReplyTokenHash(input: { + headers: Headers + recipients: ReadonlyArray +}) { + const explicit = + input.headers.get('X-Kody-Reply-Token') ?? + input.headers.get('X-Reply-Token') ?? + null + if (explicit?.trim()) { + return hashReplyToken(explicit) + } + for (const recipient of input.recipients) { + const normalized = normalizeEmailAddress(recipient) + if (!normalized) continue + const localPart = getEmailLocalPart(normalized) + const match = localPart.match(/\bkody-r-([a-f0-9]{16,128})\b/i) + if (match?.[1]) { + return hashReplyToken(match[1]) + } + } + return null +} + +export function extractReplyToken(input: { + headers: Headers + recipients: ReadonlyArray +}) { + const explicit = + input.headers.get('X-Kody-Reply-Token') ?? + input.headers.get('X-Reply-Token') ?? + null + if (explicit?.trim()) return explicit.trim() + for (const recipient of input.recipients) { + const normalized = normalizeEmailAddress(recipient) + if (!normalized) continue + const localPart = getEmailLocalPart(normalized) + const match = + localPart.match(/\bkody-r-([a-f0-9]{16,128})\b/i) ?? + localPart.match(/\+reply-([a-z0-9_-]+)/i) + if (match?.[1]) return match[1] + } + return null +} diff --git a/packages/worker/src/email/inbound.ts b/packages/worker/src/email/inbound.ts new file mode 100644 index 0000000000..28f057a08d --- /dev/null +++ b/packages/worker/src/email/inbound.ts @@ -0,0 +1,187 @@ +import { + findReplyTokenHash, + normalizeEmailAddress, + normalizeSubject, +} from './address.ts' +import { parseForwardableEmailMessage } from './parser.ts' +import { evaluateSenderPolicy } from './policy.ts' +import { + createEmailThread, + findEmailThreadForInboundMessage, + getEmailInboxById, + getEmailInboxAddressByAddress, + getEmailInboxAddressByReplyTokenHash, + insertEmailDeliveryEvent, + insertEmailMessageWithAttachments, + listEmailSenderPolicies, + touchEmailThread, +} from './repo.ts' + +export async function handleInboundEmail( + message: ForwardableEmailMessage, + env: Pick, + _ctx?: ExecutionContext, +) { + const recipient = normalizeEmailAddress(message.to) + if (!recipient) { + message.setReject('Invalid recipient address.') + return + } + + const explicitReplyTokenHash = await findReplyTokenHash({ + headers: message.headers, + recipients: [recipient], + }) + const inboxAddress = + (await getEmailInboxAddressByAddress({ + db: env.APP_DB, + address: recipient, + })) ?? + (explicitReplyTokenHash + ? await getEmailInboxAddressByReplyTokenHash({ + db: env.APP_DB, + replyTokenHash: explicitReplyTokenHash, + }) + : null) + const inbox = inboxAddress + ? await getEmailInboxById({ + db: env.APP_DB, + id: inboxAddress.inboxId, + }) + : null + + if (!inboxAddress) { + message.setReject('Unknown Kody email alias.') + return + } + if (!inbox) { + message.setReject('Email inbox is unavailable.') + return + } + if (!inbox.enabled) { + message.setReject('Email inbox is disabled.') + return + } + + const userId = inboxAddress.userId + let parsed: Awaited> + try { + parsed = await parseForwardableEmailMessage(message) + } catch (error) { + const reason = + error instanceof Error ? error.message : 'Failed to parse inbound email.' + message.setReject(reason) + await insertEmailDeliveryEvent({ + db: env.APP_DB, + userId, + inboxId: inbox.id, + eventType: 'rejected', + provider: 'cloudflare-email-routing', + detail: { + recipient, + reason, + phase: 'parse', + }, + }).catch(() => undefined) + return + } + const policies = await listEmailSenderPolicies({ + db: env.APP_DB, + userId, + inboxId: inboxAddress.inboxId, + }) + const decision = await evaluateSenderPolicy({ + envelopeFrom: parsed.envelopeFrom, + fromAddress: parsed.headerFrom, + replyToken: parsed.replyToken, + rules: policies, + defaultDecision: inbox.mode === 'accept' ? 'accepted' : 'quarantined', + }) + const policyDecision = decision.decision + const now = new Date().toISOString() + const subjectNormalized = normalizeSubject(parsed.subject) + const existingThread = await findEmailThreadForInboundMessage({ + db: env.APP_DB, + userId, + inboxId: inbox.id, + references: parsed.references, + inReplyToHeader: parsed.inReplyTo, + }) + const thread = + existingThread ?? + (await createEmailThread({ + db: env.APP_DB, + userId, + inboxId: inbox?.id ?? null, + subjectNormalized, + rootMessageIdHeader: parsed.messageId, + lastMessageAt: now, + })) + const stored = await insertEmailMessageWithAttachments({ + db: env.APP_DB, + message: { + direction: 'inbound', + userId, + inboxId: inbox?.id ?? null, + threadId: thread.id, + senderIdentityId: null, + fromAddress: parsed.headerFrom, + envelopeFrom: parsed.envelopeFrom, + toAddresses: parsed.to.map((entry) => entry.address), + ccAddresses: parsed.cc.map((entry) => entry.address), + bccAddresses: parsed.bcc.map((entry) => entry.address), + replyToAddresses: parsed.replyTo.map((entry) => entry.address), + subject: parsed.subject, + messageIdHeader: parsed.messageId, + inReplyToHeader: parsed.inReplyTo, + references: parsed.references, + headers: parsed.headers, + authResults: parsed.authResults, + textBody: parsed.textBody, + htmlBody: parsed.htmlBody, + rawMime: parsed.rawMime, + rawSize: parsed.rawSize, + policyDecision, + processingStatus: 'stored', + providerMessageId: null, + error: policyDecision === 'rejected' ? decision.reasons.join(', ') : null, + receivedAt: now, + sentAt: null, + }, + attachments: parsed.attachments.map((attachment) => ({ + filename: attachment.filename, + contentType: attachment.contentType, + contentId: attachment.contentId, + disposition: attachment.disposition, + size: attachment.size, + storageKind: 'raw-mime', + storageKey: null, + })), + }) + await touchEmailThread({ + db: env.APP_DB, + threadId: thread.id, + lastMessageAt: now, + }) + await insertEmailDeliveryEvent({ + db: env.APP_DB, + messageId: stored.id, + userId, + inboxId: inbox?.id ?? null, + eventType: + policyDecision === 'accepted' + ? 'received' + : policyDecision === 'quarantined' + ? 'quarantined' + : 'rejected', + provider: 'cloudflare-email-routing', + detail: { + recipient, + decision: policyDecision, + reasons: decision.reasons, + }, + }) + if (policyDecision === 'rejected') { + message.setReject(decision.reasons.join(', ')) + } +} diff --git a/packages/worker/src/email/inbound.workers.test.ts b/packages/worker/src/email/inbound.workers.test.ts new file mode 100644 index 0000000000..bbb8c04c8b --- /dev/null +++ b/packages/worker/src/email/inbound.workers.test.ts @@ -0,0 +1,224 @@ +import { env } from 'cloudflare:workers' +import { expect, test } from 'vitest' +import { + getEmailDomain, + getEmailLocalPart, + requireNormalizedEmailAddress, +} from './address.ts' +import { handleInboundEmail } from './inbound.ts' +import { + createEmailInbox, + createEmailInboxAddress, + createEmailThread, + listEmailMessages, + upsertEmailSenderPolicy, +} from './repo.ts' +import { ensureEmailTestSchema } from './test-schema.ts' + +function createForwardableEmailMessage(input: { + from: string + to: string + raw: string +}): ForwardableEmailMessage & { rejectedReason: string | null } { + const encoded = new TextEncoder().encode(input.raw) + const headers = new Headers() + for (const line of input.raw.split(/\r?\n/)) { + if (!line.trim()) break + const separator = line.indexOf(':') + if (separator <= 0) continue + headers.append(line.slice(0, separator), line.slice(separator + 1).trim()) + } + return { + from: input.from, + to: input.to, + headers, + raw: new Blob([encoded]).stream(), + rawSize: encoded.byteLength, + rejectedReason: null, + setReject(reason: string) { + this.rejectedReason = reason + }, + async forward() { + return { messageId: 'unused-forward' } + }, + async reply() { + return { messageId: 'unused-reply' } + }, + } +} + +test('inbound email handler stores quarantined and accepted messages by sender policy', async () => { + await ensureEmailTestSchema(env.APP_DB) + const userId = `email-user-${crypto.randomUUID()}` + const address = requireNormalizedEmailAddress( + `support-${crypto.randomUUID()}@example.com`, + ) + const inbox = await createEmailInbox({ + db: env.APP_DB, + userId, + name: 'Support', + description: 'Support inbox', + mode: 'quarantine', + }) + await createEmailInboxAddress({ + db: env.APP_DB, + inboxId: inbox.id, + userId, + address, + localPart: getEmailLocalPart(address), + domain: getEmailDomain(address), + }) + + const quarantinedMessage = createForwardableEmailMessage({ + from: 'stranger@example.net', + to: address, + raw: [ + 'From: Stranger ', + `To: ${address}`, + 'Subject: Unknown sender', + 'Message-ID: ', + '', + 'Please help.', + ].join('\r\n'), + }) + await handleInboundEmail(quarantinedMessage, env) + expect(quarantinedMessage.rejectedReason).toBeNull() + + await upsertEmailSenderPolicy({ + db: env.APP_DB, + userId, + inboxId: inbox.id, + kind: 'domain', + value: 'trusted.example', + effect: 'allow', + }) + const acceptedMessage = createForwardableEmailMessage({ + from: 'agent@trusted.example', + to: address, + raw: [ + 'From: Agent ', + `To: ${address}`, + 'Subject: Approved sender', + 'Message-ID: ', + '', + 'Approved body.', + ].join('\r\n'), + }) + await handleInboundEmail(acceptedMessage, env) + expect(acceptedMessage.rejectedReason).toBeNull() + + const messages = await listEmailMessages({ + db: env.APP_DB, + userId, + inboxId: inbox.id, + limit: 10, + }) + expect(messages.map((message) => message.policyDecision)).toEqual([ + 'accepted', + 'quarantined', + ]) + expect(messages[0]).toMatchObject({ + fromAddress: 'agent@trusted.example', + subject: 'Approved sender', + processingStatus: 'stored', + }) + expect(messages[1]).toMatchObject({ + fromAddress: 'stranger@example.net', + subject: 'Unknown sender', + error: null, + }) + + const normalizedExistingThread = await createEmailThread({ + db: env.APP_DB, + userId, + inboxId: inbox.id, + subjectNormalized: 'normalized subject', + }) + const subjectOnlyMessage = createForwardableEmailMessage({ + from: 'agent@trusted.example', + to: address, + raw: [ + 'From: Agent ', + `To: ${address}`, + 'Subject: Re: Normalized Subject', + '', + 'Subject-only body.', + ].join('\r\n'), + }) + await handleInboundEmail(subjectOnlyMessage, env) + const subjectOnly = await listEmailMessages({ + db: env.APP_DB, + userId, + inboxId: inbox.id, + limit: 1, + }) + expect(subjectOnly[0]?.threadId).not.toBe(normalizedExistingThread.id) +}) + +test('inbound email handler rejects unknown aliases without persisting them', async () => { + await ensureEmailTestSchema(env.APP_DB) + const recipient = `missing-${crypto.randomUUID()}@example.com` + const message = createForwardableEmailMessage({ + from: 'stranger@example.net', + to: recipient, + raw: [ + 'From: Stranger ', + `To: ${recipient}`, + 'Subject: Unknown alias', + '', + 'Please help.', + ].join('\r\n'), + }) + + await handleInboundEmail(message, env) + + expect(message.rejectedReason).toBe('Unknown Kody email alias.') + const messages = await listEmailMessages({ + db: env.APP_DB, + userId: 'unknown', + limit: 10, + }) + expect(messages).toEqual([]) +}) + +test('inbound email handler rejects malformed messages without persisting them', async () => { + await ensureEmailTestSchema(env.APP_DB) + const userId = `email-parse-user-${crypto.randomUUID()}` + const address = requireNormalizedEmailAddress( + `parse-${crypto.randomUUID()}@example.com`, + ) + const inbox = await createEmailInbox({ + db: env.APP_DB, + userId, + name: 'Parse failures', + description: 'Parse failure inbox', + mode: 'quarantine', + }) + await createEmailInboxAddress({ + db: env.APP_DB, + inboxId: inbox.id, + userId, + address, + localPart: getEmailLocalPart(address), + domain: getEmailDomain(address), + }) + const message = createForwardableEmailMessage({ + from: 'sender@example.net', + to: address, + raw: 'Subject: Too large\r\n\r\nBody', + }) + Object.defineProperty(message, 'rawSize', { + value: 600 * 1024, + }) + + await handleInboundEmail(message, env) + + expect(message.rejectedReason).toMatch(/too large/) + const messages = await listEmailMessages({ + db: env.APP_DB, + userId, + inboxId: inbox.id, + limit: 10, + }) + expect(messages).toEqual([]) +}) diff --git a/packages/worker/src/email/outbound.ts b/packages/worker/src/email/outbound.ts new file mode 100644 index 0000000000..3aa0cdfc1b --- /dev/null +++ b/packages/worker/src/email/outbound.ts @@ -0,0 +1,341 @@ +import { sendCloudflareEmail } from '#app/email/cloudflare-email.ts' +import { normalizeEmailAddress } from './address.ts' +import { + createEmailThread, + getEmailMessageById, + getEmailMessageByMessageIdHeader, + getVerifiedSenderIdentity, + insertEmailMessage, + insertEmailDeliveryEvent, + updateEmailMessageDelivery, +} from './repo.ts' +import { + type EmailMessageRecord, + type EmailProcessingStatus, +} from './types.ts' + +type SendEmailEnv = Pick< + Env, + | 'APP_DB' + | 'EMAIL' + | 'CLOUDFLARE_ACCOUNT_ID' + | 'CLOUDFLARE_API_BASE_URL' + | 'CLOUDFLARE_API_TOKEN' +> + +export type EmailSendInput = { + env: SendEmailEnv + userId: string + from: string + to: string | Array + subject: string + text?: string | null + html?: string | null + replyTo?: string | null + inReplyToHeader?: string | null + references?: Array + threadId?: string | null + inboxId?: string | null +} + +export type EmailSendResult = { + message: EmailMessageRecord + providerMessageId: string | null + status: EmailProcessingStatus + error: string | null +} + +function normalizeRecipientList(to: string | Array) { + const values = Array.isArray(to) ? to : [to] + const normalized = values + .map(normalizeEmailAddress) + .filter((value): value is string => typeof value === 'string') + if (normalized.length === 0) { + throw new Error('At least one recipient email address is required.') + } + return Array.from(new Set(normalized)) +} + +function buildHeaders(input: { + messageId: string + inReplyTo?: string | null + references?: Array +}) { + const headers: Record = { + 'Message-ID': input.messageId, + 'X-Kody-Email-Message-Id': input.messageId, + } + if (input.inReplyTo) headers['In-Reply-To'] = input.inReplyTo + if (input.references && input.references.length > 0) { + headers['References'] = input.references.join(' ') + } + return headers +} + +async function requireStoredEmailMessage(input: { + env: SendEmailEnv + userId: string + messageId: string +}) { + const stored = await getEmailMessageById({ + db: input.env.APP_DB, + userId: input.userId, + messageId: input.messageId, + }) + if (!stored) { + throw new Error(`Email message disappeared after delivery update: ${input.messageId}`) + } + return stored +} + +async function sendViaBinding(input: { + env: SendEmailEnv + from: string + to: Array + subject: string + text?: string | null + html?: string | null + replyTo?: string | null + headers: Record +}) { + const binding = input.env.EMAIL + if (!binding) return { sent: false, messageId: null } + const result = await binding.send({ + from: input.from, + to: input.to, + subject: input.subject, + ...(input.replyTo ? { replyTo: input.replyTo } : {}), + headers: input.headers, + ...(input.text ? { text: input.text } : {}), + ...(input.html ? { html: input.html } : {}), + }) + return { sent: true, messageId: result.messageId ?? null } +} + +async function sendViaRestFallback(input: { + env: SendEmailEnv + from: string + to: Array + subject: string + text?: string | null + html?: string | null + replyTo?: string | null + headers: Record +}) { + const result = await sendCloudflareEmail( + { + accountId: input.env.CLOUDFLARE_ACCOUNT_ID, + apiBaseUrl: input.env.CLOUDFLARE_API_BASE_URL, + apiToken: input.env.CLOUDFLARE_API_TOKEN, + }, + { + from: input.from, + to: input.to.length === 1 ? input.to[0]! : input.to, + subject: input.subject, + html: input.html ?? input.text ?? input.subject, + text: input.text ?? undefined, + replyTo: input.replyTo ?? undefined, + headers: Object.keys(input.headers).length > 0 ? input.headers : undefined, + }, + ) + if (!result.ok) { + throw new Error(result.error ?? 'Cloudflare email send was skipped.') + } + if (!input.html && !input.text) { + throw new Error('Email text or HTML body is required.') + } + return result.messageId ?? null +} + +export async function sendOutboundEmail( + input: EmailSendInput, +): Promise { + const from = normalizeEmailAddress(input.from) + if (!from) throw new Error('A valid from email address is required.') + const senderIdentity = await getVerifiedSenderIdentity({ + db: input.env.APP_DB, + userId: input.userId, + email: from, + }) + if (!senderIdentity || senderIdentity.status !== 'verified') { + throw new Error(`Sender identity is not verified: ${from}`) + } + const original = input.inReplyToHeader + ? await getEmailMessageByMessageIdHeader({ + db: input.env.APP_DB, + userId: input.userId, + messageIdHeader: input.inReplyToHeader, + }) + : null + if (input.inReplyToHeader) { + if (!original) { + throw new Error( + `Cannot reply because original message ${input.inReplyToHeader} was not found.`, + ) + } + } + + const to = normalizeRecipientList(input.to) + const subject = input.subject.trim() + if (!subject) throw new Error('Email subject is required.') + 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 existingThreadId = original?.threadId ?? input.threadId ?? null + const thread = existingThreadId + ? null + : await createEmailThread({ + db: input.env.APP_DB, + userId: input.userId, + inboxId: original?.inboxId ?? input.inboxId ?? null, + subjectNormalized: subject.toLowerCase(), + rootMessageIdHeader: input.inReplyToHeader ?? null, + lastMessageAt: new Date().toISOString(), + }) + const threadId = existingThreadId ?? thread?.id ?? null + const messageIdHeader = `<${crypto.randomUUID()}@kody.local>` + const sendHeaders = buildHeaders({ + messageId: messageIdHeader, + inReplyTo: input.inReplyToHeader ?? null, + references: input.references ?? [], + }) + const message = await insertEmailMessage({ + db: input.env.APP_DB, + message: { + direction: 'outbound', + userId: input.userId, + inboxId: original?.inboxId ?? input.inboxId ?? null, + threadId, + senderIdentityId: senderIdentity.id, + fromAddress: from, + envelopeFrom: from, + toAddresses: to, + ccAddresses: [], + bccAddresses: [], + replyToAddresses: input.replyTo + ? [normalizeEmailAddress(input.replyTo)].filter( + (value): value is string => typeof value === 'string', + ) + : [], + subject, + messageIdHeader, + inReplyToHeader: input.inReplyToHeader ?? null, + references: input.references ?? [], + headers: sendHeaders, + authResults: null, + textBody: text, + htmlBody: html, + rawMime: null, + rawSize: null, + policyDecision: 'accepted', + processingStatus: 'stored', + providerMessageId: null, + error: null, + receivedAt: null, + sentAt: null, + }, + }) + await insertEmailDeliveryEvent({ + db: input.env.APP_DB, + messageId: message.id, + userId: input.userId, + inboxId: null, + eventType: 'send_requested', + provider: 'cloudflare-email', + providerMessageId: null, + detail: { to, from, subject }, + }) + + try { + const bindingResult = await sendViaBinding({ + env: input.env, + from, + to, + subject, + text, + html, + replyTo: input.replyTo + ? (normalizeEmailAddress(input.replyTo) ?? undefined) + : undefined, + headers: sendHeaders, + }) + const providerMessageId = bindingResult.sent + ? bindingResult.messageId + : await sendViaRestFallback({ + env: input.env, + from, + to, + subject, + text, + html, + replyTo: input.replyTo + ? (normalizeEmailAddress(input.replyTo) ?? undefined) + : undefined, + headers: sendHeaders, + }) + await updateEmailMessageDelivery({ + db: input.env.APP_DB, + messageId: message.id, + status: 'sent', + providerMessageId, + error: null, + sentAt: new Date().toISOString(), + }) + await insertEmailDeliveryEvent({ + db: input.env.APP_DB, + messageId: message.id, + userId: input.userId, + inboxId: null, + eventType: 'sent', + provider: 'cloudflare-email', + providerMessageId, + detail: { providerMessageId }, + }) + return { + message: await requireStoredEmailMessage({ + env: input.env, + userId: input.userId, + messageId: message.id, + }), + providerMessageId, + status: 'sent', + error: null, + } + } catch (error) { + const messageText = error instanceof Error ? error.message : String(error) + await updateEmailMessageDelivery({ + db: input.env.APP_DB, + messageId: message.id, + status: 'failed', + providerMessageId: null, + error: messageText, + sentAt: null, + }).catch((updateError) => { + console.warn('email-delivery-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-delivery-failure-event-insert-failed', eventError) + }) + return { + message: + (await getEmailMessageById({ + db: input.env.APP_DB, + userId: input.userId, + messageId: message.id, + })) ?? message, + providerMessageId: null, + status: 'failed', + error: messageText, + } + } +} diff --git a/packages/worker/src/email/outbound.workers.test.ts b/packages/worker/src/email/outbound.workers.test.ts new file mode 100644 index 0000000000..99c870d7f2 --- /dev/null +++ b/packages/worker/src/email/outbound.workers.test.ts @@ -0,0 +1,195 @@ +import { env } from 'cloudflare:workers' +import { expect, test } from 'vitest' +import { getEmailDomain } from './address.ts' +import { ensureEmailTestSchema } from './test-schema.ts' +import { + getEmailMessageById, + listEmailMessages, + upsertEmailSenderIdentity, +} from './repo.ts' +import { sendOutboundEmail } from './outbound.ts' + +test('sendOutboundEmail uses SendEmail binding and stores sent delivery state', async () => { + await ensureEmailTestSchema(env.APP_DB) + const userId = `email-outbound-user-${crypto.randomUUID()}` + const from = `sender-${crypto.randomUUID()}@example.com` + const sent: Array = [] + const sendEnv = { + ...env, + EMAIL: { + async send(message: EmailMessage) { + sent.push(message) + return { messageId: 'provider-message-123' } + }, + }, + } + await upsertEmailSenderIdentity({ + db: env.APP_DB, + userId, + email: from, + domain: getEmailDomain(from), + status: 'verified', + }) + + const result = await sendOutboundEmail({ + env: sendEnv, + userId, + from, + to: 'recipient@example.com', + subject: 'Hello from Kody', + text: 'Body', + }) + + expect(sent).toHaveLength(1) + expect(result.status).toBe('sent') + expect(result.providerMessageId).toBe('provider-message-123') + const stored = await getEmailMessageById({ + db: env.APP_DB, + userId, + messageId: result.message.id, + }) + expect(stored).toMatchObject({ + direction: 'outbound', + processingStatus: 'sent', + providerMessageId: 'provider-message-123', + fromAddress: from, + }) + const listed = await listEmailMessages({ + db: env.APP_DB, + userId, + direction: 'outbound', + limit: 5, + }) + expect(listed.map((message) => message.id)).toContain(result.message.id) +}) + +test('sendOutboundEmail does not fall back when binding sends without message id', async () => { + await ensureEmailTestSchema(env.APP_DB) + const userId = `email-outbound-null-id-user-${crypto.randomUUID()}` + const from = `sender-${crypto.randomUUID()}@example.com` + let bindingSendCount = 0 + const originalFetch = globalThis.fetch + globalThis.fetch = (async () => { + throw new Error('REST fallback should not be called') + }) as typeof fetch + try { + await upsertEmailSenderIdentity({ + db: env.APP_DB, + userId, + email: from, + domain: getEmailDomain(from), + status: 'verified', + }) + const result = await sendOutboundEmail({ + env: { + ...env, + EMAIL: { + async send() { + bindingSendCount += 1 + return { messageId: null as unknown as string } + }, + }, + }, + userId, + from, + to: 'recipient@example.com', + subject: 'No provider id', + text: 'Body', + }) + expect(bindingSendCount).toBe(1) + expect(result).toMatchObject({ + status: 'sent', + providerMessageId: null, + error: null, + }) + } finally { + globalThis.fetch = originalFetch + } +}) + +test('sendOutboundEmail preserves reply headers and records failed fallback sends', async () => { + await ensureEmailTestSchema(env.APP_DB) + const userId = `email-outbound-fallback-user-${crypto.randomUUID()}` + const from = `sender-${crypto.randomUUID()}@example.com` + const fetchCalls: Array<{ url: string; body: Record }> = [] + const originalFetch = globalThis.fetch + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + fetchCalls.push({ + url: String(input), + body: JSON.parse(String(init?.body ?? '{}')) as Record, + }) + return new Response( + JSON.stringify({ + success: false, + errors: [{ message: 'provider down' }], + }), + { status: 500, headers: { 'content-type': 'application/json' } }, + ) + }) as typeof fetch + + try { + await upsertEmailSenderIdentity({ + db: env.APP_DB, + userId, + email: from, + domain: getEmailDomain(from), + status: 'verified', + }) + const original = await sendOutboundEmail({ + env: { + ...env, + EMAIL: { + async send() { + return { messageId: 'original-provider-message' } + }, + }, + }, + userId, + from, + to: 'recipient@example.com', + subject: 'Hello from Kody', + text: 'Original body', + }) + + const result = await sendOutboundEmail({ + env: { + ...env, + 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, + from, + to: 'recipient@example.com', + subject: 'Re: Hello from Kody', + text: 'Body', + replyTo: 'reply@example.com', + inReplyToHeader: original.message.messageIdHeader, + references: [''], + }) + + expect(result.status).toBe('failed') + expect(result.error).toBe('provider down') + expect(fetchCalls).toHaveLength(1) + expect(fetchCalls[0]?.body).toMatchObject({ + replyTo: 'reply@example.com', + headers: { + 'In-Reply-To': original.message.messageIdHeader, + References: '', + }, + }) + const stored = await getEmailMessageById({ + db: env.APP_DB, + userId, + messageId: result.message.id, + }) + expect(stored).toMatchObject({ + processingStatus: 'failed', + error: 'provider down', + }) + } finally { + globalThis.fetch = originalFetch + } +}) + diff --git a/packages/worker/src/email/parser.node.test.ts b/packages/worker/src/email/parser.node.test.ts new file mode 100644 index 0000000000..b780c0c4f4 --- /dev/null +++ b/packages/worker/src/email/parser.node.test.ts @@ -0,0 +1,124 @@ +import { expect, test } from 'vitest' +import { parseForwardableEmailMessage } from './parser.ts' + +function createMessage(raw: string, to = 'support@example.com'): ForwardableEmailMessage { + const headers = new Headers() + for (const line of raw.split(/\r?\n/)) { + if (!line) break + const separator = line.indexOf(':') + if (separator > 0) { + headers.append(line.slice(0, separator), line.slice(separator + 1).trim()) + } + } + return { + from: 'sender@example.com', + to, + headers, + raw: new Response(raw).body!, + rawSize: new TextEncoder().encode(raw).byteLength, + setReject() {}, + async forward() { + return { messageId: 'forwarded' } + }, + async reply() { + return { messageId: 'reply' } + }, + } satisfies ForwardableEmailMessage +} + +test('parseForwardableEmailMessage extracts headers, bodies, and attachment metadata', async () => { + const raw = [ + 'From: Sender ', + 'To: Support ', + 'Subject: Hello', + 'Message-ID: ', + 'Content-Type: multipart/mixed; boundary="b"', + '', + '--b', + 'Content-Type: text/plain; charset=utf-8', + '', + 'Plain body', + '--b', + 'Content-Type: text/plain; name="note.txt"', + 'Content-Disposition: attachment; filename="note.txt"', + '', + 'Attachment body', + '--b--', + '', + ].join('\r\n') + + const parsed = await parseForwardableEmailMessage(createMessage(raw)) + + expect(parsed).toMatchObject({ + envelopeFrom: 'sender@example.com', + envelopeTo: 'support@example.com', + headerFrom: 'sender@example.com', + subject: 'Hello', + messageId: '', + textBody: expect.stringContaining('Plain body'), + }) + expect(parsed.to.map((entry) => entry.address)).toContain('support@example.com') + expect(parsed.attachments).toEqual([ + expect.objectContaining({ + filename: 'note.txt', + contentType: 'text/plain', + disposition: 'attachment', + size: expect.any(Number), + }), + ]) +}) + +test('parseForwardableEmailMessage deduplicates overlapping to addresses and parses alternate reply token header', async () => { + const raw = [ + 'From: Sender ', + 'To: Support ', + 'X-Reply-Token: alternate-token', + 'Subject: Hello', + '', + 'Plain body', + ].join('\r\n') + + const parsed = await parseForwardableEmailMessage(createMessage(raw)) + + expect(parsed.replyToken).toBe('alternate-token') + expect(parsed.to).toEqual([{ name: null, address: 'support@example.com' }]) +}) + +test('parseForwardableEmailMessage parses kody-r reply tokens from recipients', async () => { + const raw = [ + 'From: Sender ', + 'To: Support ', + 'Subject: Reply token', + '', + 'Body', + ].join('\r\n') + + const parsed = await parseForwardableEmailMessage( + createMessage(raw, 'kody-r-0123456789abcdef@example.com'), + ) + + expect(parsed.replyToken).toBe('0123456789abcdef') +}) + +test('parseForwardableEmailMessage rejects oversized raw MIME', async () => { + await expect( + parseForwardableEmailMessage(createMessage('Subject: Oversized\n\nbody'), { + maxRawSize: 5, + }), + ).rejects.toThrow(/too large/) +}) + +test('parseForwardableEmailMessage accepts both explicit reply token headers', async () => { + const raw = [ + 'From: Sender ', + 'To: Support ', + 'Subject: Reply token', + 'X-Reply-Token: token-123', + '', + 'Body', + ].join('\r\n') + + const parsed = await parseForwardableEmailMessage(createMessage(raw)) + + expect(parsed.replyToken).toBe('token-123') +}) diff --git a/packages/worker/src/email/parser.ts b/packages/worker/src/email/parser.ts new file mode 100644 index 0000000000..c6f31db656 --- /dev/null +++ b/packages/worker/src/email/parser.ts @@ -0,0 +1,176 @@ +import PostalMime, { + type Address as PostalAddress, + type Attachment as PostalAttachment, +} from 'postal-mime' +import { + extractReplyToken, + normalizeEmailAddress, + parseHeaderAddressList, +} from './address.ts' +import { + type EmailAttachmentMetadata, + type EmailMailbox, + type ParsedInboundEmail, +} from './types.ts' + +export const maxInlineRawMimeBytes = 512 * 1024 + +function flattenPostalAddresses( + addresses: PostalAddress | Array | undefined, +): Array { + const input = Array.isArray(addresses) ? addresses : addresses ? [addresses] : [] + const out: Array = [] + for (const address of input) { + if ('group' in address && Array.isArray(address.group)) { + for (const groupAddress of address.group) { + const normalized = normalizeEmailAddress(groupAddress.address) + if (normalized) { + out.push({ name: groupAddress.name || null, address: normalized }) + } + } + continue + } + if (!address.address) continue + const normalized = normalizeEmailAddress(address.address) + if (normalized) out.push({ name: address.name || null, address: normalized }) + } + return out +} + +function normalizeReferences(value: string | undefined) { + if (!value) return [] + return value + .split(/\s+/) + .map((entry) => entry.trim()) + .filter(Boolean) +} + +function headersToRecord( + _headers: Headers, + parsedHeaders: Array<{ name: string; value: string }>, +) { + const record: Record> = {} + for (const header of parsedHeaders) { + const key = header.name.toLowerCase() + record[key] = [...(record[key] ?? []), header.value] + } + return record +} + +function getHeader(headers: Headers, name: string) { + return headers.get(name)?.trim() || null +} + +function getPostalHeader( + headers: Array<{ key: string; value: string }>, + name: string, +) { + const normalized = name.toLowerCase() + return ( + headers.find((header) => header.key.toLowerCase() === normalized)?.value ?? + null + ) +} + +function attachmentSize(attachment: PostalAttachment) { + const content = attachment.content + if (typeof content === 'string') { + return new TextEncoder().encode(content).byteLength + } + return content.byteLength +} + +function toAttachmentMetadata( + attachments: Array, +): Array { + return attachments.map((attachment) => ({ + filename: attachment.filename, + contentType: attachment.mimeType, + contentId: attachment.contentId ?? null, + disposition: attachment.disposition, + size: attachmentSize(attachment), + })) +} + +function dedupeMailboxes(addresses: Array) { + const seen = new Set() + const out: Array = [] + for (const mailbox of addresses) { + if (seen.has(mailbox.address)) continue + seen.add(mailbox.address) + out.push(mailbox) + } + return out +} + +export async function parseForwardableEmailMessage( + message: ForwardableEmailMessage, + options: { maxRawSize?: number } = {}, +): Promise { + const maxRawSize = options.maxRawSize ?? maxInlineRawMimeBytes + if (message.rawSize > maxRawSize) { + throw new Error( + `Inbound email raw MIME is too large (${message.rawSize} bytes, max ${maxRawSize}).`, + ) + } + const rawMime = await new Response(message.raw).text() + const parsed = await PostalMime.parse(rawMime, { + attachmentEncoding: 'arraybuffer', + }) + const parsedHeaders = parsed.headers.map((header) => ({ + name: header.originalKey || header.key, + value: header.value, + })) + const headerFrom = flattenPostalAddresses(parsed.from) + const headerFromFallback = parseHeaderAddressList( + getHeader(message.headers, 'From'), + )[0] + const fromAddress = + headerFrom[0]?.address ?? + headerFromFallback?.address ?? + normalizeEmailAddress(message.from) ?? + null + const normalizedEnvelopeTo = normalizeEmailAddress(message.to) + const toAddresses = dedupeMailboxes([ + ...(normalizedEnvelopeTo + ? [{ name: null, address: normalizedEnvelopeTo }] + : []), + ...flattenPostalAddresses(parsed.to), + ...parseHeaderAddressList(getHeader(message.headers, 'To')), + ]) + const ccAddresses = dedupeMailboxes(flattenPostalAddresses(parsed.cc)) + const bccAddresses = dedupeMailboxes(flattenPostalAddresses(parsed.bcc)) + const replyToAddresses = dedupeMailboxes(flattenPostalAddresses(parsed.replyTo)) + const subject = parsed.subject ?? getHeader(message.headers, 'Subject') + const messageIdHeader = + parsed.messageId ?? getPostalHeader(parsed.headers, 'message-id') + const inReplyToHeader = + parsed.inReplyTo ?? getPostalHeader(parsed.headers, 'in-reply-to') + return { + envelopeFrom: normalizeEmailAddress(message.from) ?? message.from, + envelopeTo: normalizeEmailAddress(message.to) ?? message.to, + headerFrom: fromAddress, + to: toAddresses, + cc: ccAddresses, + bcc: bccAddresses, + replyTo: replyToAddresses, + subject, + messageId: messageIdHeader ?? null, + inReplyTo: inReplyToHeader ?? null, + references: normalizeReferences(parsed.references), + headers: headersToRecord(message.headers, parsedHeaders), + authResults: + getHeader(message.headers, 'Authentication-Results') ?? + getHeader(message.headers, 'ARC-Authentication-Results'), + textBody: parsed.text || null, + htmlBody: typeof parsed.html === 'string' && parsed.html ? parsed.html : null, + rawMime, + rawSize: message.rawSize, + attachments: toAttachmentMetadata(parsed.attachments), + replyToken: extractReplyToken({ + headers: message.headers, + recipients: toAddresses.map((address) => address.address), + }), + } +} + diff --git a/packages/worker/src/email/policy.node.test.ts b/packages/worker/src/email/policy.node.test.ts new file mode 100644 index 0000000000..b6b9ef7889 --- /dev/null +++ b/packages/worker/src/email/policy.node.test.ts @@ -0,0 +1,70 @@ +import { expect, test } from 'vitest' +import { hashReplyToken } from './address.ts' +import { evaluateSenderPolicy } from './policy.ts' +import { type EmailSenderPolicyRecord } from './types.ts' + +function policy( + input: Partial & + Pick, +): EmailSenderPolicyRecord { + return { + id: input.id ?? `${input.kind}-${input.value}`, + userId: 'user-1', + inboxId: input.inboxId ?? null, + packageId: input.packageId ?? null, + kind: input.kind, + value: input.value, + effect: input.effect ?? 'allow', + enabled: input.enabled ?? true, + createdAt: '2026-04-30T00:00:00.000Z', + updatedAt: '2026-04-30T00:00:00.000Z', + } +} + +test('sender policy accepts exact senders, domains, and reply tokens before default quarantine', async () => { + const replyToken = 'reply-token-123' + const replyTokenHash = await hashReplyToken(replyToken) + const exact = await evaluateSenderPolicy({ + fromAddress: 'Alice@Example.com', + envelopeFrom: 'bounce@mailer.test', + rules: [policy({ kind: 'sender', value: 'alice@example.com' })], + }) + expect(exact).toMatchObject({ + decision: 'accepted', + policyKind: 'sender', + }) + + const domain = await evaluateSenderPolicy({ + fromAddress: 'sender@trusted.example', + envelopeFrom: null, + rules: [policy({ kind: 'domain', value: 'trusted.example' })], + }) + expect(domain).toMatchObject({ + decision: 'accepted', + policyKind: 'domain', + }) + + const token = await evaluateSenderPolicy({ + fromAddress: 'unknown@example.com', + envelopeFrom: null, + replyToken, + rules: [ + policy({ kind: 'sender', value: 'somebody@example.com' }), + policy({ kind: 'reply_token', value: replyTokenHash }), + ], + }) + expect(token).toMatchObject({ + decision: 'accepted', + policyKind: 'reply_token', + }) + + const quarantined = await evaluateSenderPolicy({ + fromAddress: 'unknown@example.com', + envelopeFrom: null, + rules: [policy({ kind: 'sender', value: 'somebody@example.com' })], + }) + expect(quarantined).toMatchObject({ + decision: 'quarantined', + ruleId: null, + }) +}) diff --git a/packages/worker/src/email/policy.ts b/packages/worker/src/email/policy.ts new file mode 100644 index 0000000000..dd919b083b --- /dev/null +++ b/packages/worker/src/email/policy.ts @@ -0,0 +1,127 @@ +import { + type EmailPolicyEvaluation, + type EmailSenderPolicyRecord, + type EmailPolicyDecision, + type EmailPolicyKind, +} from './types.ts' +import { hashReplyToken, normalizeEmailAddress } from './address.ts' + +type EvaluateSenderPolicyInput = { + fromAddress: string | null + envelopeFrom: string | null + replyToken?: string | null + rules: ReadonlyArray + defaultDecision?: EmailPolicyDecision +} + +function getSenderCandidates(input: { + fromAddress: string | null + envelopeFrom: string | null +}) { + return [input.fromAddress, input.envelopeFrom] + .map((value) => (value ? normalizeEmailAddress(value) : null)) + .filter((value): value is string => value !== null) +} + +function getDomain(address: string) { + const separator = address.lastIndexOf('@') + return separator === -1 ? null : address.slice(separator + 1) +} + +function decisionFromRule(rule: EmailSenderPolicyRecord): EmailPolicyDecision { + switch (rule.effect) { + case 'allow': + return 'accepted' + case 'quarantine': + return 'quarantined' + case 'reject': + return 'rejected' + default: { + const exhaustive: never = rule.effect + throw new Error(`Unsupported email policy effect: ${exhaustive}`) + } + } +} + +function matchSenderPolicyRule(input: { + rule: EmailSenderPolicyRecord + senderCandidates: ReadonlyArray + replyTokenHash: string | null +}) { + const value = input.rule.value.trim().toLowerCase() + switch (input.rule.kind) { + case 'sender': + return input.senderCandidates.some((candidate) => candidate === value) + case 'domain': + return input.senderCandidates.some((candidate) => getDomain(candidate) === value) + case 'reply_token': + return input.replyTokenHash !== null && input.replyTokenHash === value + default: { + const exhaustive: never = input.rule.kind + throw new Error(`Unsupported email policy kind: ${exhaustive}`) + } + } +} + +function createDecision(input: { + decision: EmailPolicyDecision + reason: string + rule: EmailSenderPolicyRecord | null +}): EmailPolicyEvaluation { + return { + decision: input.decision, + reasons: [input.reason], + ruleId: input.rule?.id ?? null, + policyKind: input.rule?.kind ?? null, + } +} + +function sortPolicyRules(rules: ReadonlyArray) { + const order: Record = { + reply_token: 0, + sender: 1, + domain: 2, + } + return [...rules] + .filter((rule) => rule.enabled) + .sort((left, right) => order[left.kind] - order[right.kind]) +} + +export async function evaluateSenderPolicy( + input: EvaluateSenderPolicyInput, +): Promise { + const senderCandidates = getSenderCandidates(input) + const replyTokenHash = input.replyToken + ? await hashReplyToken(input.replyToken) + : null + for (const rule of sortPolicyRules(input.rules)) { + if ( + matchSenderPolicyRule({ + rule, + senderCandidates, + replyTokenHash, + }) + ) { + return createDecision({ + decision: decisionFromRule(rule), + reason: `Matched ${rule.kind} policy.`, + rule, + }) + } + } + const fallback = input.defaultDecision ?? 'quarantined' + switch (fallback) { + case 'accepted': + case 'quarantined': + case 'rejected': + return createDecision({ + decision: fallback, + reason: 'No sender policy matched.', + rule: null, + }) + default: { + const exhaustive: never = fallback + throw new Error(`Unsupported default email policy decision: ${exhaustive}`) + } + } +} diff --git a/packages/worker/src/email/repo.ts b/packages/worker/src/email/repo.ts new file mode 100644 index 0000000000..2ed4660295 --- /dev/null +++ b/packages/worker/src/email/repo.ts @@ -0,0 +1,1090 @@ +import { + type EmailAttachmentRecord, + type EmailDirection, + type EmailDeliveryEventType, + type EmailInboxAddressRecord, + type EmailInboxMode, + type EmailInboxRecord, + type EmailMessageRecord, + type EmailPolicyDecision, + type EmailProcessingStatus, + type EmailSenderIdentityRecord, + type EmailPolicyEffect, + type EmailPolicyKind, + type EmailSenderPolicyRecord, + type EmailThreadRecord, +} from './types.ts' + +type D1RunResult = Awaited> + +function nowIso() { + return new Date().toISOString() +} + +function normalizeRunChanges(result: D1RunResult) { + const changes = result.meta['changes'] + return typeof changes === 'number' ? changes : Number(changes ?? 0) +} + +function parseJsonArray(value: string | null) { + if (!value) return [] + try { + const parsed = JSON.parse(value) as unknown + return Array.isArray(parsed) ? parsed : [] + } catch { + return [] + } +} + +function parseOptionalJsonRecord(value: string | null) { + if (!value) return null + try { + const parsed = JSON.parse(value) as unknown + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + return null + } + return parsed as Record + } catch { + return null + } +} + +function mapInboxRow(row: Record): EmailInboxRecord { + return { + id: String(row['id']), + userId: String(row['user_id']), + packageId: row['package_id'] == null ? null : String(row['package_id']), + name: String(row['name']), + description: row['description'] == null ? '' : String(row['description']), + mode: String(row['mode']) as EmailInboxMode, + enabled: Number(row['enabled']) === 1, + createdAt: String(row['created_at']), + updatedAt: String(row['updated_at']), + } +} + +function mapInboxAddressRow( + row: Record, +): EmailInboxAddressRecord { + return { + id: String(row['id']), + inboxId: String(row['inbox_id']), + userId: String(row['user_id']), + address: String(row['address']), + localPart: String(row['local_part']), + domain: String(row['domain']), + replyTokenHash: + row['reply_token_hash'] == null ? null : String(row['reply_token_hash']), + enabled: Number(row['enabled']) === 1, + createdAt: String(row['created_at']), + updatedAt: String(row['updated_at']), + } +} + +function mapSenderIdentityRow( + row: Record, +): EmailSenderIdentityRecord { + return { + id: String(row['id']), + userId: String(row['user_id']), + packageId: row['package_id'] == null ? null : String(row['package_id']), + email: String(row['email']), + domain: String(row['domain']), + displayName: + row['display_name'] == null ? null : String(row['display_name']), + status: String(row['status']) as EmailSenderIdentityRecord['status'], + verifiedAt: + row['verified_at'] == null ? null : String(row['verified_at']), + createdAt: String(row['created_at']), + updatedAt: String(row['updated_at']), + } +} + +function mapSenderPolicyRow( + row: Record, +): EmailSenderPolicyRecord { + return { + id: String(row['id']), + userId: String(row['user_id']), + inboxId: row['inbox_id'] == null ? null : String(row['inbox_id']), + packageId: row['package_id'] == null ? null : String(row['package_id']), + kind: String(row['kind']) as EmailPolicyKind, + value: String(row['value']), + effect: String(row['effect']) as EmailPolicyEffect, + enabled: Number(row['enabled']) === 1, + createdAt: String(row['created_at']), + updatedAt: String(row['updated_at']), + } +} + +function mapThreadRow(row: Record): EmailThreadRecord { + return { + id: String(row['id']), + userId: String(row['user_id']), + inboxId: row['inbox_id'] == null ? null : String(row['inbox_id']), + subjectNormalized: + row['subject_normalized'] == null + ? null + : String(row['subject_normalized']), + rootMessageIdHeader: + row['root_message_id_header'] == null + ? null + : String(row['root_message_id_header']), + lastMessageAt: String(row['last_message_at']), + createdAt: String(row['created_at']), + updatedAt: String(row['updated_at']), + } +} + +function mapMessageRow(row: Record): EmailMessageRecord { + return { + id: String(row['id']), + direction: String(row['direction']) as EmailDirection, + userId: String(row['user_id']), + inboxId: row['inbox_id'] == null ? null : String(row['inbox_id']), + threadId: row['thread_id'] == null ? null : String(row['thread_id']), + senderIdentityId: + row['sender_identity_id'] == null + ? null + : String(row['sender_identity_id']), + fromAddress: + row['from_address'] == null ? null : String(row['from_address']), + envelopeFrom: + row['envelope_from'] == null ? null : String(row['envelope_from']), + toAddresses: parseJsonArray(String(row['to_addresses_json'] ?? '[]')).filter( + (value): value is string => typeof value === 'string', + ), + ccAddresses: parseJsonArray(String(row['cc_addresses_json'] ?? '[]')).filter( + (value): value is string => typeof value === 'string', + ), + bccAddresses: parseJsonArray( + String(row['bcc_addresses_json'] ?? '[]'), + ).filter((value): value is string => typeof value === 'string'), + replyToAddresses: parseJsonArray( + String(row['reply_to_addresses_json'] ?? '[]'), + ).filter((value): value is string => typeof value === 'string'), + subject: row['subject'] == null ? null : String(row['subject']), + messageIdHeader: + row['message_id_header'] == null + ? null + : String(row['message_id_header']), + inReplyToHeader: + row['in_reply_to_header'] == null + ? null + : String(row['in_reply_to_header']), + references: parseJsonArray(String(row['references_json'] ?? '[]')).filter( + (value): value is string => typeof value === 'string', + ), + headers: parseOptionalJsonRecord( + row['headers_json'] == null ? null : String(row['headers_json']), + ), + authResults: + row['auth_results'] == null ? null : String(row['auth_results']), + textBody: row['text_body'] == null ? null : String(row['text_body']), + htmlBody: row['html_body'] == null ? null : String(row['html_body']), + rawMime: row['raw_mime'] == null ? null : String(row['raw_mime']), + rawSize: row['raw_size'] == null ? null : Number(row['raw_size']), + policyDecision: String(row['policy_decision']) as EmailPolicyDecision, + processingStatus: String( + row['processing_status'], + ) as EmailProcessingStatus, + providerMessageId: + row['provider_message_id'] == null + ? null + : String(row['provider_message_id']), + error: row['error'] == null ? null : String(row['error']), + receivedAt: + row['received_at'] == null ? null : String(row['received_at']), + sentAt: row['sent_at'] == null ? null : String(row['sent_at']), + createdAt: String(row['created_at']), + updatedAt: String(row['updated_at']), + } +} + +function mapAttachmentRow(row: Record): EmailAttachmentRecord { + return { + id: String(row['id']), + messageId: String(row['message_id']), + filename: row['filename'] == null ? null : String(row['filename']), + contentType: + row['content_type'] == null ? null : String(row['content_type']), + contentId: row['content_id'] == null ? null : String(row['content_id']), + disposition: + row['disposition'] == null ? null : String(row['disposition']), + size: Number(row['size'] ?? 0), + storageKind: String(row['storage_kind']), + storageKey: row['storage_key'] == null ? null : String(row['storage_key']), + createdAt: String(row['created_at']), + } +} + +export async function createEmailInbox(input: { + db: D1Database + userId: string + name: string + description?: string | null + mode: EmailInboxMode + packageId?: string | null +}) { + const timestamp = nowIso() + const row = { + id: crypto.randomUUID(), + user_id: input.userId, + package_id: input.packageId ?? null, + name: input.name, + description: input.description ?? null, + mode: input.mode, + enabled: 1, + created_at: timestamp, + updated_at: timestamp, + } + await input.db + .prepare( + `INSERT INTO email_inboxes ( + id, user_id, package_id, name, description, mode, enabled, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + row.id, + row.user_id, + row.package_id, + row.name, + row.description, + row.mode, + row.enabled, + row.created_at, + row.updated_at, + ) + .run() + return mapInboxRow(row) +} + +export async function createEmailInboxAddress(input: { + db: D1Database + inboxId: string + userId: string + address: string + localPart: string + domain: string + replyTokenHash?: string | null +}) { + const timestamp = nowIso() + const row = { + id: crypto.randomUUID(), + inbox_id: input.inboxId, + user_id: input.userId, + address: input.address, + local_part: input.localPart, + domain: input.domain, + reply_token_hash: input.replyTokenHash ?? null, + enabled: 1, + created_at: timestamp, + updated_at: timestamp, + } + await input.db + .prepare( + `INSERT INTO email_inbox_addresses ( + id, inbox_id, user_id, address, local_part, domain, reply_token_hash, enabled, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + row.id, + row.inbox_id, + row.user_id, + row.address, + row.local_part, + row.domain, + row.reply_token_hash, + row.enabled, + row.created_at, + row.updated_at, + ) + .run() + return mapInboxAddressRow(row) +} + +export async function deleteEmailInboxById(input: { + db: D1Database + inboxId: string +}) { + await input.db + .prepare( + `DELETE FROM email_inboxes + WHERE id = ?`, + ) + .bind(input.inboxId) + .run() +} + +export async function createEmailInboxWithAddress(input: { + db: D1Database + userId: string + name: string + description?: string | null + mode: EmailInboxMode + packageId?: string | null + address: string + localPart: string + domain: string + replyTokenHash?: string | null +}) { + const inbox = await createEmailInbox({ + db: input.db, + userId: input.userId, + name: input.name, + description: input.description ?? null, + mode: input.mode, + packageId: input.packageId ?? null, + }) + try { + const address = await createEmailInboxAddress({ + db: input.db, + inboxId: inbox.id, + userId: input.userId, + address: input.address, + localPart: input.localPart, + domain: input.domain, + replyTokenHash: input.replyTokenHash ?? null, + }) + return { inbox, address } + } catch (error) { + await deleteEmailInboxById({ + db: input.db, + inboxId: inbox.id, + }).catch(() => undefined) + throw error + } +} + +export async function listEmailInboxesForUser(input: { + db: D1Database + userId: string +}) { + const result = await input.db + .prepare( + `SELECT * + FROM email_inboxes + WHERE user_id = ? + ORDER BY created_at DESC, id DESC`, + ) + .bind(input.userId) + .all>() + return (result.results ?? []).map(mapInboxRow) +} + +export async function listEmailInboxAddressesForUser(input: { + db: D1Database + userId: string +}) { + const result = await input.db + .prepare( + `SELECT * + FROM email_inbox_addresses + WHERE user_id = ? + ORDER BY created_at DESC, id DESC`, + ) + .bind(input.userId) + .all>() + return (result.results ?? []).map(mapInboxAddressRow) +} + +export async function getEmailInboxAddressByAddress(input: { + db: D1Database + address: string +}) { + const row = await input.db + .prepare( + `SELECT * + FROM email_inbox_addresses + WHERE address = ? + AND enabled = 1 + LIMIT 1`, + ) + .bind(input.address) + .first>() + return row ? mapInboxAddressRow(row) : null +} + +export async function getEmailInboxAddressByReplyTokenHash(input: { + db: D1Database + replyTokenHash: string +}) { + const row = await input.db + .prepare( + `SELECT * + FROM email_inbox_addresses + WHERE reply_token_hash = ? + AND enabled = 1 + LIMIT 1`, + ) + .bind(input.replyTokenHash) + .first>() + return row ? mapInboxAddressRow(row) : null +} + +export async function getEmailInboxById(input: { + db: D1Database + userId?: string + id: string +}) { + const row = await input.db + .prepare( + `SELECT * + FROM email_inboxes + WHERE id = ? + AND (? IS NULL OR user_id = ?) + LIMIT 1`, + ) + .bind(input.id, input.userId ?? null, input.userId ?? null) + .first>() + return row ? mapInboxRow(row) : null +} + +export async function upsertEmailSenderIdentity(input: { + db: D1Database + userId: string + email: string + domain: string + displayName?: string | null + status?: EmailSenderIdentityRecord['status'] + verifiedAt?: string | null + packageId?: string | null +}) { + const existing = await input.db + .prepare( + `SELECT * + FROM email_sender_identities + WHERE user_id = ? + AND email = ? + LIMIT 1`, + ) + .bind(input.userId, input.email) + .first>() + const timestamp = nowIso() + const row = { + id: existing ? String(existing['id']) : crypto.randomUUID(), + user_id: input.userId, + package_id: input.packageId ?? null, + email: input.email, + domain: input.domain, + display_name: input.displayName ?? '', + status: input.status ?? 'verified', + verified_at: input.verifiedAt ?? timestamp, + created_at: existing ? String(existing['created_at']) : timestamp, + updated_at: timestamp, + } + await input.db + .prepare( + `INSERT INTO email_sender_identities ( + id, user_id, package_id, email, domain, display_name, status, verified_at, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(user_id, email) DO UPDATE SET + package_id = excluded.package_id, + domain = excluded.domain, + display_name = excluded.display_name, + status = excluded.status, + verified_at = excluded.verified_at, + updated_at = excluded.updated_at`, + ) + .bind( + row.id, + row.user_id, + row.package_id, + row.email, + row.domain, + row.display_name, + row.status, + row.verified_at, + row.created_at, + row.updated_at, + ) + .run() + const persisted = await input.db + .prepare( + `SELECT * + FROM email_sender_identities + WHERE user_id = ? + AND email = ? + LIMIT 1`, + ) + .bind(input.userId, input.email) + .first>() + if (!persisted) { + throw new Error('Failed to read saved sender identity.') + } + return mapSenderIdentityRow(persisted) +} + +export async function getVerifiedSenderIdentity(input: { + db: D1Database + userId: string + email: string +}) { + const row = await input.db + .prepare( + `SELECT * + FROM email_sender_identities + WHERE user_id = ? + AND email = ? + AND status = 'verified' + LIMIT 1`, + ) + .bind(input.userId, input.email) + .first>() + return row ? mapSenderIdentityRow(row) : null +} + +export async function upsertEmailSenderPolicy(input: { + db: D1Database + userId: string + kind: EmailPolicyKind + value: string + effect: EmailPolicyEffect + inboxId?: string | null + packageId?: string | null +}) { + const existing = await input.db + .prepare( + `SELECT * + FROM email_sender_policies + WHERE user_id = ? + AND kind = ? + AND value = ? + AND COALESCE(inbox_id, '') = COALESCE(?, '') + AND COALESCE(package_id, '') = COALESCE(?, '') + LIMIT 1`, + ) + .bind( + input.userId, + input.kind, + input.value, + input.inboxId ?? null, + input.packageId ?? null, + ) + .first>() + const timestamp = nowIso() + const row = { + id: existing ? String(existing['id']) : crypto.randomUUID(), + user_id: input.userId, + inbox_id: input.inboxId ?? null, + package_id: input.packageId ?? null, + kind: input.kind, + value: input.value, + effect: input.effect, + enabled: 1, + created_at: existing ? String(existing['created_at']) : timestamp, + updated_at: timestamp, + } + if (existing) { + await input.db + .prepare( + `UPDATE email_sender_policies + SET effect = ?, + enabled = 1, + updated_at = ? + WHERE id = ?`, + ) + .bind(row.effect, row.updated_at, row.id) + .run() + } else { + await input.db + .prepare( + `INSERT INTO email_sender_policies ( + id, user_id, inbox_id, package_id, kind, value, effect, enabled, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + row.id, + row.user_id, + row.inbox_id, + row.package_id, + row.kind, + row.value, + row.effect, + row.enabled, + row.created_at, + row.updated_at, + ) + .run() + } + return mapSenderPolicyRow(row) +} + +export async function disableEmailSenderPolicy(input: { + db: D1Database + userId: string + kind: EmailPolicyKind + value: string + inboxId?: string | null + packageId?: string | null +}) { + const result = await input.db + .prepare( + `UPDATE email_sender_policies + SET enabled = 0, + updated_at = ? + WHERE user_id = ? + AND kind = ? + AND value = ? + AND COALESCE(inbox_id, '') = COALESCE(?, '') + AND COALESCE(package_id, '') = COALESCE(?, '')`, + ) + .bind( + nowIso(), + input.userId, + input.kind, + input.value, + input.inboxId ?? null, + input.packageId ?? null, + ) + .run() + return normalizeRunChanges(result) > 0 +} + +export async function listEmailSenderPolicies(input: { + db: D1Database + userId: string + inboxId?: string | null + includeDisabled?: boolean +}) { + const result = await input.db + .prepare( + `SELECT * + FROM email_sender_policies + WHERE user_id = ? + AND (? IS NULL OR inbox_id IS NULL OR inbox_id = ?) + AND (? = 1 OR enabled = 1) + ORDER BY inbox_id DESC, kind ASC, value ASC`, + ) + .bind( + input.userId, + input.inboxId ?? null, + input.inboxId ?? null, + input.includeDisabled ? 1 : 0, + ) + .all>() + return (result.results ?? []).map(mapSenderPolicyRow) +} + +export async function findThreadForMessage(input: { + db: D1Database + userId: string + inboxId?: string | null + references: Array + inReplyToHeader?: string | null +}) { + const ids = [ + ...input.references, + ...(input.inReplyToHeader ? [input.inReplyToHeader] : []), + ].filter(Boolean) + for (const id of ids) { + const row = await input.db + .prepare( + `SELECT thread.* + FROM email_threads thread + JOIN email_messages message ON message.thread_id = thread.id + WHERE thread.user_id = ? + AND (? IS NULL OR thread.inbox_id = ?) + AND message.message_id_header = ? + LIMIT 1`, + ) + .bind(input.userId, input.inboxId ?? null, input.inboxId ?? null, id) + .first>() + if (row) return mapThreadRow(row) + } + return null +} + +export const findEmailThreadForInboundMessage = findThreadForMessage + +export async function createEmailThread(input: { + db: D1Database + userId: string + inboxId?: string | null + subjectNormalized?: string | null + rootMessageIdHeader?: string | null + lastMessageAt?: string | null +}) { + const timestamp = nowIso() + const row = { + id: crypto.randomUUID(), + user_id: input.userId, + inbox_id: input.inboxId ?? null, + subject_normalized: input.subjectNormalized ?? '', + root_message_id_header: input.rootMessageIdHeader ?? null, + last_message_at: input.lastMessageAt ?? timestamp, + created_at: timestamp, + updated_at: timestamp, + } + await input.db + .prepare( + `INSERT INTO email_threads ( + id, user_id, inbox_id, subject_normalized, root_message_id_header, last_message_at, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + row.id, + row.user_id, + row.inbox_id, + row.subject_normalized, + row.root_message_id_header, + row.last_message_at, + row.created_at, + row.updated_at, + ) + .run() + return mapThreadRow(row) +} + +export async function touchEmailThread(input: { + db: D1Database + threadId: string + lastMessageAt?: string | null +}) { + await input.db + .prepare( + `UPDATE email_threads + SET last_message_at = ?, + updated_at = ? + WHERE id = ?`, + ) + .bind(input.lastMessageAt ?? nowIso(), nowIso(), input.threadId) + .run() +} + +export async function insertEmailMessage(input: { + db: D1Database + message: { + id?: string + direction: EmailDirection + userId: string + inboxId?: string | null + threadId?: string | null + senderIdentityId?: string | null + fromAddress?: string | null + envelopeFrom?: string | null + toAddresses?: Array + ccAddresses?: Array + bccAddresses?: Array + replyToAddresses?: Array + subject?: string | null + messageIdHeader?: string | null + inReplyToHeader?: string | null + references?: Array + headers?: Record | null + authResults?: string | null + textBody?: string | null + htmlBody?: string | null + rawMime?: string | null + rawSize?: number | null + policyDecision: EmailPolicyDecision + processingStatus: EmailProcessingStatus + providerMessageId?: string | null + error?: string | null + receivedAt?: string | null + sentAt?: string | null + } +}) { + const timestamp = nowIso() + const row = { + id: input.message.id ?? crypto.randomUUID(), + direction: input.message.direction, + user_id: input.message.userId, + inbox_id: input.message.inboxId ?? null, + thread_id: input.message.threadId ?? null, + sender_identity_id: input.message.senderIdentityId ?? null, + from_address: input.message.fromAddress ?? '', + envelope_from: input.message.envelopeFrom ?? null, + to_addresses_json: JSON.stringify(input.message.toAddresses ?? []), + cc_addresses_json: JSON.stringify(input.message.ccAddresses ?? []), + bcc_addresses_json: JSON.stringify(input.message.bccAddresses ?? []), + reply_to_addresses_json: JSON.stringify( + input.message.replyToAddresses ?? [], + ), + subject: input.message.subject ?? '', + message_id_header: input.message.messageIdHeader ?? null, + in_reply_to_header: input.message.inReplyToHeader ?? null, + references_json: JSON.stringify(input.message.references ?? []), + headers_json: input.message.headers + ? JSON.stringify(input.message.headers) + : '{}', + auth_results: input.message.authResults ?? null, + text_body: input.message.textBody ?? null, + html_body: input.message.htmlBody ?? null, + raw_mime: input.message.rawMime ?? null, + raw_size: input.message.rawSize ?? 0, + policy_decision: input.message.policyDecision, + processing_status: input.message.processingStatus, + provider_message_id: input.message.providerMessageId ?? null, + error: input.message.error ?? null, + received_at: input.message.receivedAt ?? null, + sent_at: input.message.sentAt ?? null, + created_at: timestamp, + updated_at: timestamp, + } + await input.db + .prepare( + `INSERT INTO email_messages ( + id, direction, user_id, inbox_id, thread_id, sender_identity_id, + from_address, envelope_from, to_addresses_json, cc_addresses_json, + bcc_addresses_json, reply_to_addresses_json, subject, message_id_header, + in_reply_to_header, references_json, headers_json, auth_results, + text_body, html_body, raw_mime, raw_size, policy_decision, + processing_status, provider_message_id, error, received_at, sent_at, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + row.id, + row.direction, + row.user_id, + row.inbox_id, + row.thread_id, + row.sender_identity_id, + row.from_address, + row.envelope_from, + row.to_addresses_json, + row.cc_addresses_json, + row.bcc_addresses_json, + row.reply_to_addresses_json, + row.subject, + row.message_id_header, + row.in_reply_to_header, + row.references_json, + row.headers_json, + row.auth_results, + row.text_body, + row.html_body, + row.raw_mime, + row.raw_size, + row.policy_decision, + row.processing_status, + row.provider_message_id, + row.error, + row.received_at, + row.sent_at, + row.created_at, + row.updated_at, + ) + .run() + return mapMessageRow(row) +} + +export async function updateEmailMessageDelivery(input: { + db: D1Database + messageId: string + status: EmailProcessingStatus + providerMessageId?: string | null + error?: string | null + sentAt?: string | null +}) { + await input.db + .prepare( + `UPDATE email_messages + SET processing_status = ?, + provider_message_id = ?, + error = ?, + sent_at = ?, + updated_at = ? + WHERE id = ?`, + ) + .bind( + input.status, + input.providerMessageId ?? null, + input.error ?? null, + input.sentAt ?? null, + nowIso(), + input.messageId, + ) + .run() +} + +export async function getEmailMessageById(input: { + db: D1Database + userId: string + messageId: string +}) { + const row = await input.db + .prepare( + `SELECT * + FROM email_messages + WHERE id = ? + AND user_id = ? + LIMIT 1`, + ) + .bind(input.messageId, input.userId) + .first>() + return row ? mapMessageRow(row) : null +} + +export async function getEmailMessageByMessageIdHeader(input: { + db: D1Database + userId: string + messageIdHeader: string +}) { + const row = await input.db + .prepare( + `SELECT * + FROM email_messages + WHERE user_id = ? + AND message_id_header = ? + LIMIT 1`, + ) + .bind(input.userId, input.messageIdHeader) + .first>() + return row ? mapMessageRow(row) : null +} + +export async function listEmailMessages(input: { + db: D1Database + userId: string + inboxId?: string | null + direction?: EmailDirection | null + processingStatus?: EmailProcessingStatus | null + policyDecision?: EmailPolicyDecision | null + limit: number +}) { + const result = await input.db + .prepare( + `SELECT * + FROM email_messages + WHERE user_id = ? + AND (? IS NULL OR inbox_id = ?) + AND (? IS NULL OR direction = ?) + AND (? IS NULL OR processing_status = ?) + AND (? IS NULL OR policy_decision = ?) + ORDER BY created_at DESC, id DESC + LIMIT ?`, + ) + .bind( + input.userId, + input.inboxId ?? null, + input.inboxId ?? null, + input.direction ?? null, + input.direction ?? null, + input.processingStatus ?? null, + input.processingStatus ?? null, + input.policyDecision ?? null, + input.policyDecision ?? null, + input.limit, + ) + .all>() + return (result.results ?? []).map(mapMessageRow) +} + +export async function deleteEmailMessageById(input: { + db: D1Database + messageId: string +}) { + await input.db + .prepare(`DELETE FROM email_messages WHERE id = ?`) + .bind(input.messageId) + .run() +} + +export async function insertEmailAttachments(input: { + db: D1Database + messageId: string + attachments: Array<{ + filename?: string | null + contentType?: string | null + contentId?: string | null + disposition?: string | null + size?: number | null + storageKind: string + storageKey?: string | null + }> +}) { + 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() + } +} +export async function insertEmailMessageWithAttachments( + input: Parameters[0] & { + attachments: Parameters[0]['attachments'] + }, +) { + const message = await insertEmailMessage(input) + try { + await insertEmailAttachments({ + db: input.db, + messageId: message.id, + attachments: input.attachments, + }) + return message + } catch (error) { + await deleteEmailMessageById({ + db: input.db, + messageId: message.id, + }).catch(() => undefined) + throw error + } +} + +export async function listEmailAttachmentsForMessage(input: { + db: D1Database + messageId: string +}) { + const result = await input.db + .prepare( + `SELECT * + FROM email_attachments + WHERE message_id = ? + ORDER BY created_at ASC, id ASC`, + ) + .bind(input.messageId) + .all>() + return (result.results ?? []).map(mapAttachmentRow) +} + +export async function insertEmailDeliveryEvent(input: { + db: D1Database + messageId?: string | null + userId?: string | null + inboxId?: string | null + eventType: EmailDeliveryEventType + provider?: string | null + providerMessageId?: string | null + detail?: Record | null +}) { + await input.db + .prepare( + `INSERT INTO email_delivery_events ( + id, message_id, user_id, inbox_id, event_type, provider, + provider_message_id, detail_json, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + crypto.randomUUID(), + input.messageId ?? null, + input.userId ?? null, + input.inboxId ?? null, + input.eventType, + input.provider ?? null, + input.providerMessageId ?? null, + JSON.stringify(input.detail ?? {}), + nowIso(), + ) + .run() +} + diff --git a/packages/worker/src/email/test-schema.ts b/packages/worker/src/email/test-schema.ts new file mode 100644 index 0000000000..bfa9e845ca --- /dev/null +++ b/packages/worker/src/email/test-schema.ts @@ -0,0 +1,141 @@ +export async function ensureEmailTestSchema(db: D1Database) { + const statements = [ + `CREATE TABLE IF NOT EXISTS email_sender_identities ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + package_id TEXT, + email TEXT NOT NULL, + domain TEXT, + display_name TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL CHECK (status IN ('pending', 'verified', 'disabled')), + verified_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +);`, + `CREATE UNIQUE INDEX IF NOT EXISTS idx_email_sender_identities_user_email +ON email_sender_identities(user_id, email);`, + + `CREATE TABLE IF NOT EXISTS email_inboxes ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + package_id TEXT, + name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + mode TEXT NOT NULL CHECK (mode IN ('quarantine', 'accept')), + enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +);`, + + `CREATE TABLE IF NOT EXISTS email_inbox_addresses ( + id TEXT PRIMARY KEY, + inbox_id TEXT NOT NULL, + user_id TEXT NOT NULL, + address TEXT NOT NULL UNIQUE, + local_part TEXT NOT NULL, + domain TEXT NOT NULL, + reply_token_hash TEXT, + enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (inbox_id) REFERENCES email_inboxes(id) ON DELETE CASCADE +);`, + + `CREATE TABLE IF NOT EXISTS email_threads ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + inbox_id TEXT, + subject_normalized TEXT NOT NULL DEFAULT '', + root_message_id_header TEXT, + last_message_at TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +);`, + + `CREATE TABLE IF NOT EXISTS email_messages ( + id TEXT PRIMARY KEY, + direction TEXT NOT NULL CHECK (direction IN ('inbound', 'outbound')), + user_id TEXT NOT NULL, + inbox_id TEXT, + thread_id TEXT, + sender_identity_id TEXT, + from_address TEXT NOT NULL, + envelope_from TEXT, + to_addresses_json TEXT NOT NULL DEFAULT '[]', + cc_addresses_json TEXT NOT NULL DEFAULT '[]', + bcc_addresses_json TEXT NOT NULL DEFAULT '[]', + reply_to_addresses_json TEXT NOT NULL DEFAULT '[]', + subject TEXT NOT NULL DEFAULT '', + message_id_header TEXT, + in_reply_to_header TEXT, + references_json TEXT NOT NULL DEFAULT '[]', + headers_json TEXT NOT NULL DEFAULT '{}', + auth_results TEXT, + text_body TEXT, + html_body TEXT, + raw_mime TEXT, + raw_size INTEGER NOT NULL DEFAULT 0, + policy_decision TEXT NOT NULL CHECK (policy_decision IN ('accepted', 'quarantined', 'rejected')), + processing_status TEXT NOT NULL CHECK (processing_status IN ('stored', 'sent', 'failed', 'rejected')), + provider_message_id TEXT, + error TEXT, + received_at TEXT, + sent_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +);`, + + `CREATE TABLE IF NOT EXISTS email_attachments ( + id TEXT PRIMARY KEY, + message_id TEXT NOT NULL, + filename TEXT, + content_type TEXT NOT NULL, + content_id TEXT, + disposition TEXT, + size INTEGER NOT NULL DEFAULT 0, + storage_kind TEXT NOT NULL CHECK (storage_kind IN ('raw-mime', 'external', 'unavailable')), + storage_key TEXT, + created_at TEXT NOT NULL +);`, + + `CREATE TABLE IF NOT EXISTS email_sender_policies ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + inbox_id TEXT, + package_id TEXT, + kind TEXT NOT NULL CHECK (kind IN ('sender', 'domain', 'reply_token')), + value TEXT NOT NULL, + effect TEXT NOT NULL CHECK (effect IN ('allow', 'quarantine', 'reject')), + enabled INTEGER NOT NULL DEFAULT 1 CHECK (enabled IN (0, 1)), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +);`, + `CREATE UNIQUE INDEX IF NOT EXISTS idx_email_sender_policies_user_kind_value_inbox_null_package_null +ON email_sender_policies(user_id, kind, value) +WHERE inbox_id IS NULL AND package_id IS NULL;`, + `CREATE UNIQUE INDEX IF NOT EXISTS idx_email_sender_policies_user_kind_value_inbox_package_null +ON email_sender_policies(user_id, kind, value, inbox_id) +WHERE inbox_id IS NOT NULL AND package_id IS NULL;`, + `CREATE UNIQUE INDEX IF NOT EXISTS idx_email_sender_policies_user_kind_value_package_inbox_null +ON email_sender_policies(user_id, kind, value, package_id) +WHERE inbox_id IS NULL AND package_id IS NOT NULL;`, + `CREATE UNIQUE INDEX IF NOT EXISTS idx_email_sender_policies_user_kind_value_inbox_package +ON email_sender_policies(user_id, kind, value, inbox_id, package_id) +WHERE inbox_id IS NOT NULL AND package_id IS NOT NULL;`, + + `CREATE TABLE IF NOT EXISTS email_delivery_events ( + id TEXT PRIMARY KEY, + message_id TEXT, + user_id TEXT, + inbox_id TEXT, + event_type TEXT NOT NULL CHECK (event_type IN ('receive_started', 'received', 'quarantined', 'rejected', 'send_requested', 'sent', 'failed', 'policy_matched')), + provider TEXT NOT NULL DEFAULT 'kody', + provider_message_id TEXT, + detail_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL +);`, + ] + for (const statement of statements) { + await db.prepare(statement).run() + } +} diff --git a/packages/worker/src/email/types.ts b/packages/worker/src/email/types.ts new file mode 100644 index 0000000000..091324449e --- /dev/null +++ b/packages/worker/src/email/types.ts @@ -0,0 +1,204 @@ +export const emailInboxModeValues = ['quarantine', 'accept'] as const +export type EmailInboxMode = (typeof emailInboxModeValues)[number] + +export const emailPolicyKindValues = [ + 'sender', + 'domain', + 'reply_token', +] as const +export type EmailPolicyKind = (typeof emailPolicyKindValues)[number] + +export const emailPolicyEffectValues = ['allow', 'quarantine', 'reject'] as const +export type EmailPolicyEffect = (typeof emailPolicyEffectValues)[number] + +export const emailPolicyDecisionValues = [ + 'accepted', + 'quarantined', + 'rejected', +] as const +export type EmailPolicyDecision = (typeof emailPolicyDecisionValues)[number] + +export const emailDirectionValues = ['inbound', 'outbound'] as const +export type EmailDirection = (typeof emailDirectionValues)[number] + +export const emailProcessingStatusValues = [ + 'stored', + 'sent', + 'failed', + 'rejected', +] as const +export type EmailProcessingStatus = (typeof emailProcessingStatusValues)[number] + +export const emailDeliveryEventTypeValues = [ + 'receive_started', + 'received', + 'quarantined', + 'rejected', + 'send_requested', + 'sent', + 'failed', + 'policy_matched', +] as const +export type EmailDeliveryEventType = (typeof emailDeliveryEventTypeValues)[number] + +export type EmailPolicyEvaluation = { + decision: EmailPolicyDecision + reasons: Array + ruleId: string | null + policyKind: EmailPolicyKind | null +} + +export type EmailMailbox = { + name: string | null + address: string +} + +export type EmailAttachmentMetadata = { + filename: string | null + contentType: string + contentId: string | null + disposition: string | null + size: number +} + +export type ParsedInboundEmail = { + envelopeFrom: string + envelopeTo: string + headerFrom: string | null + to: Array + cc: Array + bcc: Array + replyTo: Array + subject: string | null + messageId: string | null + inReplyTo: string | null + references: Array + headers: Record> + authResults: string | null + textBody: string | null + htmlBody: string | null + rawMime: string + rawSize: number + attachments: Array + replyToken: string | null +} + +export type EmailInboxRecord = { + id: string + userId: string + packageId: string | null + name: string + description: string + mode: EmailInboxMode + enabled: boolean + createdAt: string + updatedAt: string +} + +export type EmailInboxAddressRecord = { + id: string + inboxId: string + userId: string + address: string + localPart: string + domain: string + replyTokenHash: string | null + enabled: boolean + createdAt: string + updatedAt: string +} + +export type EmailSenderIdentityRecord = { + id: string + userId: string + packageId: string | null + email: string + domain: string + displayName: string | null + status: 'pending' | 'verified' | 'disabled' + verifiedAt: string | null + createdAt: string + updatedAt: string +} + +export type EmailSenderPolicyRecord = { + id: string + userId: string + inboxId: string | null + packageId: string | null + kind: EmailPolicyKind + value: string + effect: EmailPolicyEffect + enabled: boolean + createdAt: string + updatedAt: string +} + +export type EmailThreadRecord = { + id: string + userId: string + inboxId: string | null + subjectNormalized: string | null + rootMessageIdHeader: string | null + lastMessageAt: string | null + createdAt: string + updatedAt: string +} + +export type EmailMessageRecord = { + id: string + direction: EmailDirection + userId: string + inboxId: string | null + threadId: string | null + senderIdentityId: string | null + fromAddress: string | null + envelopeFrom: string | null + toAddresses: Array + ccAddresses: Array + bccAddresses: Array + replyToAddresses: Array + subject: string | null + messageIdHeader: string | null + inReplyToHeader: string | null + references: Array + headers: Record | null + authResults: string | null + textBody: string | null + htmlBody: string | null + rawMime: string | null + rawSize: number | null + policyDecision: EmailPolicyDecision + processingStatus: EmailProcessingStatus + providerMessageId: string | null + error: string | null + receivedAt: string | null + sentAt: string | null + createdAt: string + updatedAt: string +} + +export type EmailAttachmentRecord = { + id: string + messageId: string + filename: string | null + contentType: string | null + contentId: string | null + disposition: string | null + size: number + storageKind: string + storageKey: string | null + createdAt: string +} + +export type EmailDeliveryEventRecord = { + id: string + messageId: string | null + userId: string | null + inboxId: string | null + eventType: EmailDeliveryEventType + provider: string | null + providerMessageId: string | null + detailJson: string + createdAt: string +} diff --git a/packages/worker/src/env-schema.ts b/packages/worker/src/env-schema.ts index d4cad4e5eb..10c65f34d6 100644 --- a/packages/worker/src/env-schema.ts +++ b/packages/worker/src/env-schema.ts @@ -14,6 +14,13 @@ const d1DatabaseSchema = createSchema((value, context) => { return fail('Missing APP_DB binding for database access.', context.path) }) +const optionalSendEmailSchema = createSchema( + (value, _context) => { + if (value === undefined) return { value: undefined } + return { value: value as SendEmail } + }, +) + const optionalNonEmptyStringSchema = createSchema( (value, context) => { if (value === undefined) return { value: undefined } @@ -221,6 +228,7 @@ export const EnvSchema = object({ APP_BASE_URL: optionalUrlStringSchema, APP_COMMIT_SHA: optionalCommitShaSchema, CLOUDFLARE_EMAIL_FROM: optionalNonEmptyStringSchema, + EMAIL: optionalSendEmailSchema, AI_MODE: optionalAiModeSchema, AI_MODEL: optionalNonEmptyStringSchema, AI_GATEWAY_ID: optionalNonEmptyStringSchema, diff --git a/packages/worker/src/index.ts b/packages/worker/src/index.ts index d070f6fb66..9a738fa2ab 100644 --- a/packages/worker/src/index.ts +++ b/packages/worker/src/index.ts @@ -48,6 +48,7 @@ import { } from './remote-connector/connector-session-key.ts' import { handlePackageAppRequest } from '#app/handlers/package-app.ts' import { PackageAppRuntimeBridge } from '#worker/package-runtime/package-app.ts' +import { handleInboundEmail } from '#worker/email/inbound.ts' export { ChatAgent, @@ -311,6 +312,13 @@ const workerHandler = { } return oauthProvider.fetch(request, env, ctx) }, + async email( + message: ForwardableEmailMessage, + env: Env, + ctx: ExecutionContext, + ) { + await handleInboundEmail(message, env, ctx) + }, } satisfies ExportedHandler export default Sentry.withSentry( diff --git a/packages/worker/src/mcp/capabilities/builtin-domains.ts b/packages/worker/src/mcp/capabilities/builtin-domains.ts index 8382c65090..7c8e7c28b9 100644 --- a/packages/worker/src/mcp/capabilities/builtin-domains.ts +++ b/packages/worker/src/mcp/capabilities/builtin-domains.ts @@ -1,5 +1,6 @@ import { appsDomain } from './apps/domain.ts' import { codingDomain } from './coding/domain.ts' +import { emailDomain } from './email/domain.ts' import { jobsDomain } from './jobs/domain.ts' import { metaDomain } from './meta/domain.ts' import { packagesDomain } from './packages/domain.ts' @@ -20,6 +21,7 @@ import { valuesDomain } from './values/domain.ts' export const builtinDomains = [ appsDomain, codingDomain, + emailDomain, jobsDomain, metaDomain, packagesDomain, diff --git a/packages/worker/src/mcp/capabilities/domain-metadata.ts b/packages/worker/src/mcp/capabilities/domain-metadata.ts index f9b7d4c2cb..03226f6036 100644 --- a/packages/worker/src/mcp/capabilities/domain-metadata.ts +++ b/packages/worker/src/mcp/capabilities/domain-metadata.ts @@ -2,6 +2,7 @@ export const capabilityDomainNames = { apps: 'apps', coding: 'coding', + email: 'email', home: 'home', jobs: 'jobs', math: 'math', diff --git a/packages/worker/src/mcp/capabilities/email/domain.ts b/packages/worker/src/mcp/capabilities/email/domain.ts new file mode 100644 index 0000000000..06f32b9fc2 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/domain.ts @@ -0,0 +1,29 @@ +import { defineDomain } from '#mcp/capabilities/define-domain.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { emailInboxCreateCapability } from './email-inbox-create.ts' +import { emailInboxListCapability } from './email-inbox-list.ts' +import { emailMessageGetCapability } from './email-message-get.ts' +import { emailMessageListCapability } from './email-message-list.ts' +import { emailPolicyGetCapability } from './email-policy-get.ts' +import { emailReplyCapability } from './email-reply.ts' +import { emailSendCapability } from './email-send.ts' +import { emailSenderApproveCapability } from './email-sender-approve.ts' +import { emailSenderRevokeCapability } from './email-sender-revoke.ts' + +export const emailDomain = defineDomain({ + name: capabilityDomainNames.email, + description: + 'Cloudflare-backed email primitives for creating inbox aliases, sending verified outbound mail, storing inbound messages, and enforcing sender policies.', + keywords: ['email', 'mail', 'inbox', 'quarantine', 'sender policy'], + capabilities: [ + emailInboxCreateCapability, + emailInboxListCapability, + emailMessageListCapability, + emailMessageGetCapability, + emailSendCapability, + emailReplyCapability, + emailSenderApproveCapability, + emailSenderRevokeCapability, + emailPolicyGetCapability, + ], +}) diff --git a/packages/worker/src/mcp/capabilities/email/email-domain.node.test.ts b/packages/worker/src/mcp/capabilities/email/email-domain.node.test.ts new file mode 100644 index 0000000000..867277c2ce --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/email-domain.node.test.ts @@ -0,0 +1,53 @@ +import { expect, test } from 'vitest' +import { createMcpCallerContext } from '#mcp/context.ts' +import { emailDomain } from './domain.ts' + +function createContext() { + return { + env: {} as Env, + callerContext: createMcpCallerContext({ + baseUrl: 'https://example.com', + user: { + userId: 'user-1', + email: 'user@example.com', + displayName: 'User Example', + }, + }), + } +} + +test('email domain exposes storage, sender policy, and send capabilities', () => { + expect(emailDomain.name).toBe('email') + expect(emailDomain.capabilities.map((capability) => capability.name)).toEqual( + expect.arrayContaining([ + 'email_inbox_create', + 'email_inbox_list', + 'email_message_list', + 'email_message_get', + 'email_send', + 'email_reply', + 'email_sender_approve', + 'email_sender_revoke', + 'email_policy_get', + ]), + ) +}) + +test('email capabilities require a signed-in user context', async () => { + const capability = emailDomain.capabilities.find( + (candidate) => candidate.name === 'email_message_list', + ) + if (!capability) throw new Error('email_message_list capability missing') + + await expect( + capability.handler( + { limit: 1 }, + { + ...createContext(), + callerContext: createMcpCallerContext({ + baseUrl: 'https://example.com', + }), + }, + ), + ).rejects.toThrow(/Authenticated MCP user/) +}) diff --git a/packages/worker/src/mcp/capabilities/email/email-inbox-create.ts b/packages/worker/src/mcp/capabilities/email/email-inbox-create.ts new file mode 100644 index 0000000000..a29026c391 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/email-inbox-create.ts @@ -0,0 +1,69 @@ +import { z } from 'zod' +import { + createReplyToken, + getEmailDomain, + getEmailLocalPart, + hashReplyToken, + requireNormalizedEmailAddress, +} from '#worker/email/address.ts' +import { createEmailInboxWithAddress } from '#worker/email/repo.ts' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { + emailInboxCreateOutputSchema, + emailInboxModeSchema, +} from './shared.ts' + +export const emailInboxCreateCapability = defineDomainCapability( + capabilityDomainNames.email, + { + name: 'email_inbox_create', + description: + 'Create a storage-only email inbox and routable alias for the signed-in user. Unknown senders are quarantined by default.', + keywords: ['email', 'inbox', 'alias', 'routing', 'quarantine'], + readOnly: false, + idempotent: false, + destructive: false, + inputSchema: z.object({ + name: z.string().trim().min(1), + address: z.string().email(), + description: z.string().optional(), + mode: emailInboxModeSchema.default('quarantine'), + }), + outputSchema: emailInboxCreateOutputSchema, + async handler(args, ctx) { + const user = requireMcpUser(ctx.callerContext) + const address = requireNormalizedEmailAddress(args.address, 'Inbox address') + const replyToken = createReplyToken() + const { inbox, address: alias } = await createEmailInboxWithAddress({ + db: ctx.env.APP_DB, + userId: user.userId, + name: args.name.trim(), + description: args.description?.trim() ?? '', + mode: args.mode, + address, + localPart: getEmailLocalPart(address), + domain: getEmailDomain(address), + replyTokenHash: await hashReplyToken(replyToken), + }) + return { + id: inbox.id, + name: inbox.name, + description: inbox.description, + mode: inbox.mode, + enabled: inbox.enabled, + addresses: [{ + id: alias.id, + address: alias.address, + reply_token_hash: alias.replyTokenHash, + reply_token: replyToken, + enabled: alias.enabled, + created_at: alias.createdAt, + }], + created_at: inbox.createdAt, + updated_at: inbox.updatedAt, + } + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/email/email-inbox-list.ts b/packages/worker/src/mcp/capabilities/email/email-inbox-list.ts new file mode 100644 index 0000000000..29d0ad6766 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/email-inbox-list.ts @@ -0,0 +1,54 @@ +import { z } from 'zod' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { + listEmailInboxAddressesForUser, + listEmailInboxesForUser, +} from '#worker/email/repo.ts' +import { emailInboxListSchema } from './shared.ts' + +export const emailInboxListCapability = defineDomainCapability( + capabilityDomainNames.email, + { + name: 'email_inbox_list', + description: + 'List email inboxes and routable aliases owned by the signed-in user.', + keywords: ['email', 'inbox', 'alias', 'list'], + readOnly: true, + idempotent: true, + destructive: false, + inputSchema: z.object({}), + outputSchema: emailInboxListSchema, + async handler(_args, ctx) { + const user = requireMcpUser(ctx.callerContext) + const [inboxes, addresses] = await Promise.all([ + listEmailInboxesForUser({ db: ctx.env.APP_DB, userId: user.userId }), + listEmailInboxAddressesForUser({ + db: ctx.env.APP_DB, + userId: user.userId, + }), + ]) + return { + inboxes: inboxes.map((inbox) => ({ + id: inbox.id, + name: inbox.name, + description: inbox.description, + mode: inbox.mode, + enabled: inbox.enabled, + addresses: addresses + .filter((address) => address.inboxId === inbox.id) + .map((address) => ({ + id: address.id, + address: address.address, + reply_token_hash: address.replyTokenHash, + enabled: address.enabled, + created_at: address.createdAt, + })), + created_at: inbox.createdAt, + updated_at: inbox.updatedAt, + })), + } + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/email/email-message-get.ts b/packages/worker/src/mcp/capabilities/email/email-message-get.ts new file mode 100644 index 0000000000..64ab205b19 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/email-message-get.ts @@ -0,0 +1,42 @@ +import { z } from 'zod' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { + getEmailMessageById, + listEmailAttachmentsForMessage, +} from '#worker/email/repo.ts' +import { emailMessageDetailSchema, toMessageDetail } from './shared.ts' + +export const emailMessageGetCapability = defineDomainCapability( + capabilityDomainNames.email, + { + name: 'email_message_get', + description: + 'Get one stored email message, including parsed bodies, headers, attachment metadata, and processing state.', + keywords: ['email', 'message', 'get', 'headers', 'attachments'], + readOnly: true, + idempotent: true, + destructive: false, + inputSchema: z.object({ + message_id: z.string().min(1), + }), + outputSchema: emailMessageDetailSchema, + async handler(args, ctx) { + const user = requireMcpUser(ctx.callerContext) + const message = await getEmailMessageById({ + db: ctx.env.APP_DB, + userId: user.userId, + messageId: args.message_id, + }) + if (!message) { + throw new Error(`Email message not found: ${args.message_id}`) + } + const attachments = await listEmailAttachmentsForMessage({ + db: ctx.env.APP_DB, + messageId: message.id, + }) + return toMessageDetail(message, attachments) + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/email/email-message-list.ts b/packages/worker/src/mcp/capabilities/email/email-message-list.ts new file mode 100644 index 0000000000..652a674a5c --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/email-message-list.ts @@ -0,0 +1,44 @@ +import { z } from 'zod' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { listEmailMessages } from '#worker/email/repo.ts' +import { emailMessageSummarySchema, toMessageSummary } from './shared.ts' + +export const emailMessageListCapability = defineDomainCapability( + capabilityDomainNames.email, + { + name: 'email_message_list', + description: + 'List stored inbound, quarantined, and outbound email messages owned by the signed-in user.', + keywords: ['email', 'message', 'inbox', 'quarantine', 'list'], + readOnly: true, + idempotent: true, + destructive: false, + inputSchema: z.object({ + inbox_id: z.string().min(1).optional(), + direction: z.enum(['inbound', 'outbound']).optional(), + processing_status: z.enum(['stored', 'sent', 'failed', 'rejected']).optional(), + policy_decision: z + .enum(['accepted', 'quarantined', 'rejected']) + .optional(), + limit: z.number().int().positive().max(100).default(25), + }), + outputSchema: z.object({ + messages: z.array(emailMessageSummarySchema), + }), + async handler(args, ctx) { + const user = requireMcpUser(ctx.callerContext) + const messages = await listEmailMessages({ + db: ctx.env.APP_DB, + userId: user.userId, + inboxId: args.inbox_id ?? null, + direction: args.direction ?? null, + processingStatus: args.processing_status ?? null, + policyDecision: args.policy_decision ?? null, + limit: args.limit, + }) + return { messages: messages.map(toMessageSummary) } + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/email/email-policy-get.ts b/packages/worker/src/mcp/capabilities/email/email-policy-get.ts new file mode 100644 index 0000000000..cc2a3f004c --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/email-policy-get.ts @@ -0,0 +1,48 @@ +import { z } from 'zod' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { listEmailSenderPolicies } from '#worker/email/repo.ts' +import { emailPolicySchema } from './shared.ts' + +export const emailPolicyGetCapability = defineDomainCapability( + capabilityDomainNames.email, + { + name: 'email_policy_get', + description: + 'Inspect sender allow/quarantine/reject policies for the signed-in user, optionally scoped to an inbox.', + keywords: ['email', 'policy', 'allowlist', 'quarantine', 'sender'], + readOnly: true, + idempotent: true, + destructive: false, + inputSchema: z.object({ + inbox_id: z.string().min(1).optional(), + include_disabled: z.boolean().default(false), + }), + outputSchema: z.object({ + policies: z.array(emailPolicySchema), + }), + async handler(args, ctx) { + const user = requireMcpUser(ctx.callerContext) + const policies = await listEmailSenderPolicies({ + db: ctx.env.APP_DB, + userId: user.userId, + inboxId: args.inbox_id ?? null, + includeDisabled: args.include_disabled, + }) + return { + policies: policies.map((policy) => ({ + id: policy.id, + inbox_id: policy.inboxId, + package_id: policy.packageId, + kind: policy.kind, + value: policy.value, + effect: policy.effect, + enabled: policy.enabled, + created_at: policy.createdAt, + updated_at: policy.updatedAt, + })), + } + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/email/email-reply.ts b/packages/worker/src/mcp/capabilities/email/email-reply.ts new file mode 100644 index 0000000000..70f0c1b258 --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/email-reply.ts @@ -0,0 +1,65 @@ +import { z } from 'zod' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { sendOutboundEmail } from '#worker/email/outbound.ts' +import { getEmailMessageById } from '#worker/email/repo.ts' +import { emailMessageSummarySchema, stringArray, toMessageSummary } from './shared.ts' + +export const emailReplyCapability = defineDomainCapability( + capabilityDomainNames.email, + { + name: 'email_reply', + description: + 'Reply to a stored inbound email using a verified sender identity, preserving thread headers.', + keywords: ['email', 'reply', 'thread', 'message'], + readOnly: false, + idempotent: false, + destructive: false, + inputSchema: z + .object({ + message_id: z.string().min(1), + from: z.string().email(), + text: z.string().min(1).optional(), + html: z.string().min(1).optional(), + }) + .refine((value) => value.text !== undefined || value.html !== undefined, { + message: 'Email text or HTML body is required.', + path: ['text'], + }), + outputSchema: emailMessageSummarySchema, + async handler(args, ctx) { + const user = requireMcpUser(ctx.callerContext) + const original = await getEmailMessageById({ + db: ctx.env.APP_DB, + userId: user.userId, + messageId: args.message_id, + }) + if (!original) throw new Error(`Email message not found: ${args.message_id}`) + const fromAddress = + stringArray(original.replyToAddresses)[0] ?? + original.fromAddress ?? + original.envelopeFrom + if (!fromAddress) throw new Error('Original message has no reply address.') + const result = await sendOutboundEmail({ + env: ctx.env, + userId: user.userId, + from: args.from, + to: fromAddress, + subject: original.subject?.toLowerCase().startsWith('re:') + ? original.subject + : `Re: ${original.subject ?? '(no subject)'}`, + text: args.text ?? null, + html: args.html ?? null, + inReplyToHeader: original.messageIdHeader ?? null, + references: [ + ...stringArray(original.references), + ...(original.messageIdHeader ? [original.messageIdHeader] : []), + ], + threadId: original.threadId, + inboxId: original.inboxId, + }) + return toMessageSummary(result.message) + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/email/email-send.ts b/packages/worker/src/mcp/capabilities/email/email-send.ts new file mode 100644 index 0000000000..21b7e27dad --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/email-send.ts @@ -0,0 +1,59 @@ +import { z } from 'zod' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { sendOutboundEmail } from '#worker/email/outbound.ts' +import { emailMessageSummarySchema, toMessageSummary } from './shared.ts' + +const emailSendInputSchema = z + .object({ + from: z.string().min(1), + to: z.union([z.string().min(1), z.array(z.string().min(1)).min(1)]), + subject: z.string().min(1), + text: z.string().min(1).optional(), + html: z.string().min(1).optional(), + reply_to: z.string().min(1).optional(), + }) + .refine((value) => value.text || value.html, { + message: 'Email text or HTML body is required.', + path: ['text'], + }) + +export const emailSendCapability = defineDomainCapability( + capabilityDomainNames.email, + { + name: 'email_send', + description: + 'Send an outbound email from a verified sender identity and store delivery audit events.', + keywords: ['email', 'send', 'mail', 'outbound', 'sender identity'], + readOnly: false, + idempotent: false, + destructive: false, + inputSchema: emailSendInputSchema, + outputSchema: z.object({ + message: emailMessageSummarySchema, + provider_message_id: z.string().nullable(), + status: z.string(), + error: z.string().nullable(), + }), + async handler(args, ctx) { + const user = requireMcpUser(ctx.callerContext) + const result = await sendOutboundEmail({ + env: ctx.env, + userId: user.userId, + from: args.from, + to: args.to, + subject: args.subject, + text: args.text ?? null, + html: args.html ?? null, + replyTo: args.reply_to ?? null, + }) + return { + message: toMessageSummary(result.message), + provider_message_id: result.providerMessageId, + status: result.status, + error: result.error, + } + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/email/email-sender-approve.ts b/packages/worker/src/mcp/capabilities/email/email-sender-approve.ts new file mode 100644 index 0000000000..555b168c7b --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/email-sender-approve.ts @@ -0,0 +1,84 @@ +import { z } from 'zod' +import { + getEmailDomain, + normalizeEmailAddress, + requireNormalizedEmailAddress, +} from '#worker/email/address.ts' +import { + upsertEmailSenderIdentity, + upsertEmailSenderPolicy, +} from '#worker/email/repo.ts' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { senderApprovalSchema } from './shared.ts' + +export const emailSenderApproveCapability = defineDomainCapability( + capabilityDomainNames.email, + { + name: 'email_sender_approve', + description: + 'Approve an outbound sender identity or allow an inbound sender/domain for a Kody email inbox.', + keywords: ['email', 'sender', 'approve', 'allowlist', 'identity'], + readOnly: false, + idempotent: true, + destructive: false, + inputSchema: z.object({ + email: z.string().min(1).optional(), + domain: z.string().min(1).optional(), + inbox_id: z.string().min(1).optional(), + display_name: z.string().optional(), + mode: z + .enum(['sender_identity', 'allow_sender', 'allow_domain']) + .default('allow_sender'), + }), + outputSchema: senderApprovalSchema, + async handler(args, ctx) { + const user = requireMcpUser(ctx.callerContext) + if (args.mode === 'sender_identity') { + if (!args.email) throw new Error('email is required for sender identity approval.') + const email = requireNormalizedEmailAddress(args.email) + const identity = await upsertEmailSenderIdentity({ + db: ctx.env.APP_DB, + userId: user.userId, + email, + domain: getEmailDomain(email), + displayName: args.display_name ?? null, + status: 'verified', + }) + return { + identity_id: identity.id, + policy: null, + } + } + const kind = args.mode === 'allow_domain' ? 'domain' : 'sender' + const value = + kind === 'domain' + ? (args.domain ?? (args.email ? getEmailDomain(args.email) : '')).trim().toLowerCase() + : (normalizeEmailAddress(args.email ?? '') ?? '') + if (!value) throw new Error(`${kind} approval requires a valid value.`) + const policy = await upsertEmailSenderPolicy({ + db: ctx.env.APP_DB, + userId: user.userId, + inboxId: args.inbox_id ?? null, + kind, + value, + effect: 'allow', + }) + return { + identity_id: null, + policy: { + id: policy.id, + inbox_id: policy.inboxId, + package_id: policy.packageId, + kind: policy.kind, + value: policy.value, + effect: policy.effect, + enabled: policy.enabled, + created_at: policy.createdAt, + updated_at: policy.updatedAt, + }, + } + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/email/email-sender-revoke.ts b/packages/worker/src/mcp/capabilities/email/email-sender-revoke.ts new file mode 100644 index 0000000000..b0ff78682d --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/email-sender-revoke.ts @@ -0,0 +1,54 @@ +import { z } from 'zod' +import { normalizeEmailAddress } from '#worker/email/address.ts' +import { disableEmailSenderPolicy } from '#worker/email/repo.ts' +import { defineDomainCapability } from '#mcp/capabilities/define-domain-capability.ts' +import { capabilityDomainNames } from '#mcp/capabilities/domain-metadata.ts' +import { requireMcpUser } from '#mcp/capabilities/meta/require-user.ts' +import { emailPolicyKindSchema, revokePolicyOutputSchema } from './shared.ts' + +export const emailSenderRevokeCapability = defineDomainCapability( + capabilityDomainNames.email, + { + name: 'email_sender_revoke', + description: + 'Revoke a sender, domain, or reply-token allow/quarantine/reject policy for the signed-in user.', + keywords: ['email', 'sender', 'revoke', 'policy', 'allowlist'], + readOnly: false, + idempotent: true, + destructive: false, + inputSchema: z.object({ + kind: emailPolicyKindSchema.describe('Policy kind to revoke.'), + value: z.string().min(1).describe('Policy value to revoke.'), + inbox_id: z + .string() + .min(1) + .optional() + .describe('Optional inbox id for inbox-scoped policy.'), + package_id: z + .string() + .min(1) + .optional() + .describe('Optional package id for package-scoped policy.'), + }), + outputSchema: revokePolicyOutputSchema, + async handler(args, ctx) { + const user = requireMcpUser(ctx.callerContext) + const normalizedValue = + args.kind === 'sender' + ? normalizeEmailAddress(args.value) + : args.value.trim().toLowerCase() + if (!normalizedValue) { + throw new Error('Policy value must be a valid sender or domain.') + } + const revoked = await disableEmailSenderPolicy({ + db: ctx.env.APP_DB, + userId: user.userId, + kind: args.kind, + value: normalizedValue, + inboxId: args.inbox_id ?? null, + packageId: args.package_id ?? null, + }) + return { revoked } + }, + }, +) diff --git a/packages/worker/src/mcp/capabilities/email/shared.ts b/packages/worker/src/mcp/capabilities/email/shared.ts new file mode 100644 index 0000000000..deb7851e8f --- /dev/null +++ b/packages/worker/src/mcp/capabilities/email/shared.ts @@ -0,0 +1,193 @@ +import { z } from 'zod' +import { + type EmailMessageRecord, + emailDirectionValues, + emailInboxModeValues, + emailPolicyDecisionValues, + emailPolicyEffectValues, + emailPolicyKindValues, + emailProcessingStatusValues, +} from '#worker/email/types.ts' + +export const emailInboxModeSchema = z.enum(emailInboxModeValues) +export const emailPolicyKindSchema = z.enum(emailPolicyKindValues) + +export const emailInboxSchema = z.object({ + id: z.string(), + name: z.string(), + description: z.string().nullable(), + mode: z.enum(emailInboxModeValues), + enabled: z.boolean(), + addresses: z.array( + z.object({ + id: z.string(), + address: z.string(), + reply_token_hash: z.string().nullable(), + enabled: z.boolean(), + created_at: z.string(), + }), + ), + created_at: z.string(), + updated_at: z.string(), +}) + +export const emailInboxListSchema = z.object({ + inboxes: z.array(emailInboxSchema), +}) + +export const emailInboxCreateOutputSchema = emailInboxSchema.extend({ + addresses: z.array( + z.object({ + id: z.string(), + address: z.string(), + reply_token_hash: z.string().nullable(), + reply_token: z.string(), + enabled: z.boolean(), + created_at: z.string(), + }), + ), +}) + +export const emailMessageSummarySchema = z.object({ + id: z.string(), + direction: z.enum(emailDirectionValues), + inbox_id: z.string().nullable(), + thread_id: z.string().nullable(), + from_address: z.string().nullable(), + envelope_from: z.string().nullable(), + to_addresses: z.array(z.string()), + subject: z.string().nullable(), + message_id_header: z.string().nullable(), + policy_decision: z.enum(emailPolicyDecisionValues), + processing_status: z.enum(emailProcessingStatusValues), + provider_message_id: z.string().nullable(), + error: z.string().nullable(), + received_at: z.string().nullable(), + sent_at: z.string().nullable(), + created_at: z.string(), + updated_at: z.string(), +}) + +export const emailAttachmentSchema = z.object({ + id: z.string(), + filename: z.string().nullable(), + content_type: z.string().nullable(), + content_id: z.string().nullable(), + disposition: z.string().nullable(), + size: z.number().nullable(), + storage_kind: z.string(), + storage_key: z.string().nullable(), + created_at: z.string(), +}) + +export const emailMessageDetailSchema = emailMessageSummarySchema.extend({ + cc_addresses: z.array(z.string()), + bcc_addresses: z.array(z.string()), + reply_to_addresses: z.array(z.string()), + in_reply_to_header: z.string().nullable(), + references: z.array(z.string()), + headers: z.record(z.string(), z.unknown()).nullable(), + auth_results: z.string().nullable(), + text_body: z.string().nullable(), + html_body: z.string().nullable(), + raw_size: z.number().nullable(), + attachments: z.array(emailAttachmentSchema), +}) + +export const emailPolicySchema = z.object({ + id: z.string(), + inbox_id: z.string().nullable(), + package_id: z.string().nullable(), + kind: z.enum(emailPolicyKindValues), + value: z.string(), + effect: z.enum(emailPolicyEffectValues), + enabled: z.boolean(), + created_at: z.string(), + updated_at: z.string(), +}) + +export const senderApprovalSchema = z.object({ + identity_id: z.string().nullable(), + policy: emailPolicySchema.nullable(), +}) + +export const revokePolicyOutputSchema = z.object({ + revoked: z.boolean(), +}) + +export function stringArray(values: ReadonlyArray) { + return values.filter((value): value is string => typeof value === 'string') +} + +export function toMessageSummary(message: EmailMessageRecord) { + return { + id: message.id, + direction: message.direction, + inbox_id: message.inboxId, + thread_id: message.threadId, + from_address: message.fromAddress, + envelope_from: message.envelopeFrom, + to_addresses: stringArray(message.toAddresses), + subject: message.subject, + message_id_header: message.messageIdHeader, + policy_decision: message.policyDecision, + processing_status: message.processingStatus, + provider_message_id: message.providerMessageId, + error: message.error, + received_at: message.receivedAt, + sent_at: message.sentAt, + created_at: message.createdAt, + updated_at: message.updatedAt, + } +} + +export function toMessageDetail( + message: Parameters[0] & { + ccAddresses: Array + bccAddresses: Array + replyToAddresses: Array + inReplyToHeader: string | null + references: Array + headers: Record | null + authResults: string | null + textBody: string | null + htmlBody: string | null + rawSize: number | null + }, + attachments: Array<{ + id: string + filename: string | null + contentType: string | null + contentId: string | null + disposition: string | null + size: number + storageKind: string + storageKey: string | null + createdAt: string + }>, +) { + return { + ...toMessageSummary(message), + cc_addresses: stringArray(message.ccAddresses), + bcc_addresses: stringArray(message.bccAddresses), + reply_to_addresses: stringArray(message.replyToAddresses), + in_reply_to_header: message.inReplyToHeader, + references: stringArray(message.references), + headers: message.headers, + auth_results: message.authResults, + text_body: message.textBody, + html_body: message.htmlBody, + raw_size: message.rawSize, + attachments: attachments.map((attachment) => ({ + id: attachment.id, + filename: attachment.filename, + content_type: attachment.contentType, + content_id: attachment.contentId, + disposition: attachment.disposition, + size: attachment.size, + storage_kind: attachment.storageKind, + storage_key: attachment.storageKey, + created_at: attachment.createdAt, + })), + } +} diff --git a/packages/worker/src/package-registry/types.ts b/packages/worker/src/package-registry/types.ts index 6c5f4cde6c..35a6496ce8 100644 --- a/packages/worker/src/package-registry/types.ts +++ b/packages/worker/src/package-registry/types.ts @@ -87,6 +87,25 @@ export type PackageRetrieverDefinition = z.infer< typeof packageRetrieverDefinitionSchema > +export const packageEmailInboxDefinitionSchema = z.object({ + id: z.string().regex(kodyPackageIdPattern), + mode: z.enum(['quarantine', 'accept']).optional(), +}) + +export const packageEmailHandlerDefinitionSchema = z.object({ + inbox: z.string().regex(kodyPackageIdPattern), + export: z.string().min(1), +}) + +export const packageEmailDefinitionSchema = z.object({ + inboxes: z.array(packageEmailInboxDefinitionSchema).optional(), + handlers: z.array(packageEmailHandlerDefinitionSchema).optional(), +}) + +export type PackageEmailDefinition = z.infer< + typeof packageEmailDefinitionSchema +> + const packageExportConditionSchema = z .object({ import: z.string().min(1).optional(), @@ -133,6 +152,7 @@ export const authoredPackageKodySchema = z.object({ packageRetrieverDefinitionSchema, ) .optional(), + email: packageEmailDefinitionSchema.optional(), }) export type AuthoredPackageKody = z.infer diff --git a/packages/worker/src/package-runtime/module-graph.ts b/packages/worker/src/package-runtime/module-graph.ts index af647aad50..f5d4daf630 100644 --- a/packages/worker/src/package-runtime/module-graph.ts +++ b/packages/worker/src/package-runtime/module-graph.ts @@ -105,6 +105,7 @@ export const packageContext = runtime.packageContext ?? null; export const serviceContext = runtime.serviceContext ?? null; export const service = runtime.service ?? null; export const packageSecrets = runtime.packageSecrets ?? null; +export const email = runtime.email ?? null; export default runtime; `.trim() diff --git a/packages/worker/worker-configuration.d.ts b/packages/worker/worker-configuration.d.ts index 9a78db5e41..a169de5fc3 100644 --- a/packages/worker/worker-configuration.d.ts +++ b/packages/worker/worker-configuration.d.ts @@ -1,16 +1,17 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types ./packages/worker/worker-configuration.d.ts --config packages/worker/wrangler.jsonc --env production` (hash: b2114a21543e2157af9f1648c525aa5f) -// Runtime types generated with workerd@1.20260415.1 2026-04-13 global_fetch_strictly_public,nodejs_compat +// Generated by Wrangler by running `wrangler types ./packages/worker/worker-configuration.d.ts --config packages/worker/wrangler.jsonc --env production` (hash: 78947ff6dd1fc1b92a2289332320e9fd) +// Runtime types generated with workerd@1.20260415.1 2026-04-16 global_fetch_strictly_public,nodejs_compat declare namespace Cloudflare { interface GlobalProps { mainModule: typeof import("./src/index"); - durableNamespaces: "MCP" | "ChatAgent" | "HomeConnectorSession" | "HomeMCP" | "JobManager" | "StorageRunner" | "AgentTurnRunner" | "RepoSession" | "PackageServiceInstance"; + durableNamespaces: "MCP" | "ChatAgent" | "HomeConnectorSession" | "HomeMCP" | "JobManager" | "StorageRunner" | "AgentTurnRunner" | "RepoSession" | "PackageRealtimeSession" | "PackageServiceInstance"; } interface Env { OAUTH_KV: KVNamespace; BUNDLE_ARTIFACTS_KV: KVNamespace; APP_DB: D1Database; CAPABILITY_VECTOR_INDEX: VectorizeIndex; + EMAIL: SendEmail; LOADER: WorkerLoader; APP_LOADER: WorkerLoader; AI: Ai; @@ -29,6 +30,7 @@ declare namespace Cloudflare { STORAGE_RUNNER: DurableObjectNamespace; AGENT_TURN_RUNNER: DurableObjectNamespace; REPO_SESSION: DurableObjectNamespace; + PACKAGE_REALTIME_SESSION: DurableObjectNamespace; PACKAGE_SERVICE_INSTANCE: DurableObjectNamespace; } } @@ -9508,7 +9510,6 @@ interface AiModels { "@cf/deepgram/aura-1": Base_Ai_Cf_Deepgram_Aura_1; "@cf/ai4bharat/indictrans2-en-indic-1B": Base_Ai_Cf_Ai4Bharat_Indictrans2_En_Indic_1B; "@cf/aisingapore/gemma-sea-lion-v4-27b-it": Base_Ai_Cf_Aisingapore_Gemma_Sea_Lion_V4_27B_It; - "@cf/google/gemma-4-26b-a4b-it": Base_Ai_Cf_Google_Gemma_4_26B_A4B_IT; "@cf/pfnet/plamo-embedding-1b": Base_Ai_Cf_Pfnet_Plamo_Embedding_1B; "@cf/deepgram/flux": Base_Ai_Cf_Deepgram_Flux; "@cf/deepgram/aura-2-en": Base_Ai_Cf_Deepgram_Aura_2_En; @@ -9580,16 +9581,6 @@ interface InferenceUpstreamError extends Error { interface AiInternalError extends Error { } type AiModelListType = Record; -type ModelsWithRequests = { - [K in keyof M]: Extract extends never ? never : K; -}[keyof M]; -type ModelsWithStream = { - [K in keyof M]: Extract extends never ? never : K; -}[keyof M]; type AiAsyncBatchResponse = { request_id: string; }; @@ -9610,9 +9601,9 @@ declare abstract class Ai { */ autorag(autoragId: string): AutoRAG; // Batch request - run>(model: Name, inputs: Extract, options: AiOptions & { + run(model: Name, inputs: { + requests: AiModelList[Name]['inputs'][]; + }, options: AiOptions & { queueRequest: true; }): Promise; // Raw response @@ -9624,9 +9615,7 @@ declare abstract class Ai { websocket: true; }): Promise; // Streaming - run>(model: Name, inputs: Extract & { + run(model: Name, inputs: AiModelList[Name]['inputs'] & { stream: true; }, options?: AiOptions): Promise; // Normal (default) - known model diff --git a/packages/worker/wrangler.jsonc b/packages/worker/wrangler.jsonc index 1afd08b0ff..2888e3cc66 100644 --- a/packages/worker/wrangler.jsonc +++ b/packages/worker/wrangler.jsonc @@ -1,7 +1,7 @@ { "$schema": "../../node_modules/wrangler/config-schema.json", "name": "kody", - "compatibility_date": "2026-04-13", + "compatibility_date": "2026-04-16", "compatibility_flags": ["nodejs_compat", "global_fetch_strictly_public"], "main": "./src/index.ts", "upload_source_maps": true, @@ -129,6 +129,11 @@ "migrations_dir": "./migrations", }, ], + "send_email": [ + { + "name": "EMAIL", + }, + ], "kv_namespaces": [ { "binding": "OAUTH_KV", @@ -218,6 +223,11 @@ "migrations_dir": "./migrations", }, ], + "send_email": [ + { + "name": "EMAIL", + }, + ], "kv_namespaces": [ { "binding": "OAUTH_KV", @@ -307,6 +317,11 @@ "migrations_dir": "./migrations", }, ], + "send_email": [ + { + "name": "EMAIL", + }, + ], "kv_namespaces": [ { "binding": "OAUTH_KV",