import { kody, packageStorage } from 'kody:runtime' import { buildEmailPrompt, type GmailMatch, type StoredAttachment, type StoredEmail, } from './email-briefing.ts' import recentThread from './thread.ts' import { aliasCandidates, isKentFrom, kentOnThisMessage, plusAliasesFromMessage, consumeNonKentThreadThrottle, threadOpenKey, threadRateKey, type PlusAliasHit, } from './plus-alias.ts' import { getSecretUseStatus, resolveBot, } from './registry.ts' import wake, { isWakeTimedOut } from './wake.ts' import { plainClone, storageGet } from './plain.ts' export type HandleEmailMessageReceivedInput = { event?: string message?: Record attachments?: unknown dryRun?: boolean synthetic?: boolean [key: string]: unknown } const TEXT_ATTACHMENT_TYPES = [ 'text/', 'message/rfc822', 'message/rfc2822', 'application/json', 'application/xml', 'text/calendar', ] const MAX_TEXT_ATTACHMENT_BYTES = 80_000 function pickDryRun(...values: unknown[]): boolean { return values.some((value) => value === true) } function asRecord(value: unknown): Record | null { return value && typeof value === 'object' ? (value as Record) : null } function unwrapEvent(input: HandleEmailMessageReceivedInput): { message: Record attachments: unknown[] } { const nested = asRecord(input.message) if (nested) { return { message: nested, attachments: Array.isArray(input.attachments) ? input.attachments : [], } } const payload = asRecord(input.payload) if (payload?.message && typeof payload.message === 'object') { return { message: payload.message as Record, attachments: Array.isArray(payload.attachments) ? payload.attachments : [], } } return { message: input as Record, attachments: [] } } function decodeBase64(data: string): string { try { const normalized = data.replace(/-/g, '+').replace(/_/g, '/') const pad = normalized.length % 4 === 0 ? '' : '='.repeat(4 - (normalized.length % 4)) return atob(normalized + pad) } catch { return '' } } function isTextAttachment(contentType: string | null | undefined, size: number | null | undefined): boolean { if (typeof size === 'number' && size > MAX_TEXT_ATTACHMENT_BYTES) return false const type = (contentType || '').toLowerCase() return TEXT_ATTACHMENT_TYPES.some((prefix) => type === prefix || type.startsWith(prefix)) } async function loadAttachmentText(attachment: StoredAttachment): Promise { if (!attachment.id) return null if (!isTextAttachment(attachment.content_type, attachment.size ?? null)) return null try { const got = await kody.emailAttachmentGet({ attachment_id: attachment.id }) const b64 = typeof got?.data_base64 === 'string' ? got.data_base64 : '' if (!b64) return null const text = decodeBase64(b64) return text.trim() ? text : null } catch { return null } } async function loadFullMessage(messageId: string): Promise { try { const full = await kody.emailMessageGet({ message_id: messageId }) return plainClone(full) as StoredEmail } catch { return null } } async function lookupGmail(_message: StoredEmail): Promise { // Intentionally skipped in-subscription: cross-package google stubs were // implicated in Workers Proxy serialization failures on this surface. return null } /** * Production `email.message.received` subscriber for Grok Bot plus aliases. * * Mail to `kentcdodds+patch@inbox.kody.codes` (and the same plus tag on * inbox.heykody.app / inbox.heykody.dev) wakes that bot with the full thread * briefing. Cold mail from strangers is ignored. Kent including the bot * (From Kent, Kent on To/Cc, or a later reply on that opened thread) wakes * it. Unaliased mail returns `{ handled: false }` so the Discord inbox * subscriber still runs. A late webhook ACK is recorded as handled * `wake-timed-out` instead of throwing. * * @param params.message - Email receipt metadata from the subscription event * @param params.dryRun - Inspect routing without waking * @returns Handler result with handled flag and bot/alias metadata * @example * import handleEmailMessageReceived from 'kody:@kentcdodds/grok-bot/handle-email-message-received' * const result = await handleEmailMessageReceived({ * dryRun: true, * message: { * id: 'msg-1', * from_address: 'kentcdodds@gmail.com', * to_addresses: ['kentcdodds+patch@inbox.kody.codes'], * subject: 'Fwd: Please look at this', * }, * }) * // => { handled: true, reason: 'woke', bot: { id, name }, dryRun: true } */ export default async function handleEmailMessageReceived( input: HandleEmailMessageReceivedInput = {}, ): Promise> { const dryRun = pickDryRun(input.dryRun) const unwrapped = unwrapEvent(plainClone(input) as HandleEmailMessageReceivedInput) const meta = plainClone(unwrapped.message) const messageId = typeof meta.id === 'string' ? meta.id.trim() : '' const aliases = plusAliasesFromMessage({ to_addresses: meta.to_addresses, cc_addresses: meta.cc_addresses, bcc_addresses: meta.bcc_addresses, }) if (!aliases.length) { return plainClone({ subscriber: 'grok-bot', handled: false, reason: 'no-plus-alias', messageId: messageId || null, }) } const full = (messageId ? await loadFullMessage(messageId) : null) ?? (meta as StoredEmail) const merged: StoredEmail = { ...full, id: full.id || messageId, to_addresses: full.to_addresses?.length ? full.to_addresses : (meta.to_addresses as string[]), cc_addresses: full.cc_addresses?.length ? full.cc_addresses : (meta.cc_addresses as string[]), subject: full.subject ?? (typeof meta.subject === 'string' ? meta.subject : null), from_address: full.from_address ?? (typeof meta.from_address === 'string' ? meta.from_address : null), envelope_from: full.envelope_from ?? (typeof meta.envelope_from === 'string' ? meta.envelope_from : null), reply_to_addresses: full.reply_to_addresses?.length ? full.reply_to_addresses : (meta.reply_to_addresses as string[] | undefined), thread_id: full.thread_id ?? (typeof meta.thread_id === 'string' ? meta.thread_id : null), } const kentFrom = isKentFrom(merged) const kentListed = kentOnThisMessage(merged) const threadId = typeof merged.thread_id === 'string' ? merged.thread_id.trim() : '' const bucket = packageStorage() let threadAlreadyOpen = false if (threadId) { try { threadAlreadyOpen = Boolean( await storageGet((key) => bucket.get(key), threadOpenKey(threadId)), ) } catch { threadAlreadyOpen = false } } if (!kentFrom && !kentListed && !threadAlreadyOpen) { return plainClone({ subscriber: 'grok-bot', handled: true, reason: 'ignored-cold-email', messageId: merged.id || messageId || null, from: merged.from_address ?? null, }) } if (threadId) { try { await bucket.set( threadOpenKey(threadId), JSON.stringify({ openedAt: new Date().toISOString(), from: merged.from_address ?? null, }), ) } catch { // packageStorage may be Proxy-broken; continue without open-marker } } if (!kentFrom && threadId && dryRun !== true) { try { const rate = await storageGet((key) => bucket.get(key), threadRateKey(threadId)) const throttled = consumeNonKentThreadThrottle(rate) try { await bucket.set(threadRateKey(threadId), JSON.stringify(throttled.state)) } catch { /* ignore */ } if (!throttled.ok) { return plainClone({ subscriber: 'grok-bot', handled: true, reason: throttled.reason, messageId: merged.id || messageId || null, from: merged.from_address ?? null, threadId, lockedUntil: throttled.lockedUntil, }) } } catch { // Skip throttle when storage is unusable. } } const eventAttachments = Array.isArray(unwrapped.attachments) ? unwrapped.attachments : [] const storedAttachments: StoredAttachment[] = ( merged.attachments?.length ? merged.attachments : eventAttachments ).map((item) => { const row = asRecord(item) ?? {} return { id: typeof row.id === 'string' ? row.id : undefined, filename: typeof row.filename === 'string' ? row.filename : null, content_type: typeof row.content_type === 'string' ? row.content_type : null, size: typeof row.size === 'number' ? row.size : null, } }) const withText: StoredAttachment[] = [] for (const attachment of storedAttachments) { const text = await loadAttachmentText(attachment) withText.push({ ...attachment, ...(text ? { text } : {}) }) } merged.attachments = withText const inboxId = typeof merged.inbox_id === 'string' && merged.inbox_id.trim() ? merged.inbox_id.trim() : undefined const [threadSnapshot, gmail] = await Promise.all([ threadId ? recentThread({ threadId, limit: 0, inbox_id: inboxId }).catch(() => ({ count: 1, capped: false, })) : Promise.resolve({ count: 0, capped: false }), lookupGmail(merged), ]) const wakes: Array> = [] const seenBots = new Set() for (const alias of aliases) { let record = null for (const candidate of aliasCandidates(alias.tag)) { record = await resolveBot(candidate) if (record) break } if (!record) { wakes.push({ alias: alias.address, tag: alias.tag, handled: false, reason: 'unknown-plus-alias', }) continue } if (seenBots.has(record.id)) continue seenBots.add(record.id) const secret = await getSecretUseStatus(record.secretName) const pendingSecret = !secret.exists const hasUrl = Boolean(record.url) if (!hasUrl || pendingSecret) { wakes.push({ alias: alias.address, tag: alias.tag, handled: true, reason: 'bot-not-registered-for-wake', bot: { id: record.id, name: record.name }, pendingSecret, hasUrl, }) continue } if (secret.needsHostApproval) { wakes.push({ alias: alias.address, tag: alias.tag, handled: true, reason: 'host-approval-required', bot: { id: record.id, name: record.name }, secretName: record.secretName, approvalUrl: secret.approvalUrl, }) continue } const prompt = buildEmailPrompt({ botName: record.name, alias, message: merged, threadCount: Math.max(threadSnapshot.count, threadId ? 1 : 0), threadCapped: threadSnapshot.capped === true, gmail, kentFrom, }) const woke = await wake({ bot: record.id, prompt, emailMessageId: merged.id, emailThreadId: merged.thread_id ?? undefined, emailAlias: alias.tag, emailAddress: alias.address, dryRun, }) wakes.push({ alias: alias.address, tag: alias.tag, handled: true, reason: isWakeTimedOut(woke) ? 'wake-timed-out' : 'woke', bot: { id: record.id, name: record.name }, wake: woke, }) } const handled = wakes.some((item) => item.handled === true) const reason = wakes.length === 1 && typeof wakes[0]?.reason === 'string' ? wakes[0].reason : handled ? 'woke' : 'unknown-plus-alias' return plainClone({ subscriber: 'grok-bot', handled, reason, messageId: merged.id || messageId || null, dryRun, aliases: aliases.map((item: PlusAliasHit) => item.address), wakes, }) }