mattrguerra/meridian-ops-boardmainPrivate on GitHub, published here in full

messages.ts

Language TypeScript26.5 KB690 lines

packages/gmail-client/src/messages.ts

import {
accessTokenForCredential,
GMAIL_SEND_SCOPE,
GmailAuthError,
type GmailCredential,
} from './oauth.js'
import {
BODY_CAP,
HTML_CAP,
collectAttachments,
decodeB64UrlBytes,
extractBody,
type MessageBody,
} from './body.js'
import { errorMessage, sleep } from '@meridian/shared'
/**
* One message, as much of it as this board is willing to hold.
*
* The list is metadata only. Sender, subject, date: the request asks Gmail for
* `format=metadata`, so no body is fetched to build this list, held in memory, or
* written to KV.
*
* Bodies are readable, and they are fetched one at a time when somebody opens a
* message (see `fetchBody`). That keeps the list small enough to serve on every
* page load and keeps seven people's mail out of the board's storage entirely:
* on-demand reading without the board becoming a copy of everyone's mailbox.
*/
export interface InboxMessage {
id: string
threadId: string
/** Which roster mailbox this was read from. */
mailbox: string
/** Lowercased address, plus-tag stripped. The matching and grouping key. */
sender: string
/** As the From header spelled it, e.g. "Upscaler Labs". Display only. */
senderName: string
subject: string
/**
* Gmail's own preview line, ≤140 chars.
*
* This is message content at rest, kept deliberately: the list is meant to
* read like Gmail's (sender, subject, preview). Bodies are still never
* stored; this is the same one-line preview Gmail itself shows under every
* subject, and it costs no extra fetches because messages.get returns it
* even at format=metadata.
*/
snippet: string
/** ISO. From Gmail's internalDate, which is authoritative over the Date header. */
date: string
/** Gmail's own labels, kept because CATEGORY_PROMOTIONS seeds the filter list. */
labels: string[]
/** Gmail's UNREAD label, lifted out because the list renders it on every row. */
unread: boolean
}
/** One mailbox's read, succeeded or failed. Never merged into a total. */
export interface MailboxRead {
mailbox: string
ok: boolean
error: string | null
messages: InboxMessage[]
/**
* Can this mailbox's credential send? Observed, never assumed.
*
* Three states, and they are not interchangeable (the prime invariant):
* true Google reported gmail.send among this token's scopes.
* false Google reported the scopes and send was not among them: a
* read-only token, which is what every mailbox is until it is
* re-minted with `pnpm inbox:auth`.
* null we do not know: the token response carried no `scope` field, or
* the read failed before there was anything to ask.
*
* Free: it comes off the token response the sync already makes. The 15-minute
* sync rewrites it, so a re-minted mailbox becomes sendable without a deploy.
*/
canSend: boolean | null
/**
* How many messages the window actually holds: counted, not estimated.
*
* Gmail's `resultSizeEstimate` sounds authoritative and is capped near the
* requested page size (with maxResults=500 it reports about 501 for any
* larger window), so it is only right when the result fits in one page,
* which is exactly when it is redundant.
*
* So this counts instead. `messages.list` returns ids only and costs one
* subrequest per 500, which is cheap enough to page to the end even when the
* board will not fetch them all: listing is how you learn the size, fetching
* is what costs. Null only when the list could not be paged to the end.
*/
windowTotal: number | null
/**
* True when the window held more than this read fetched: observed, by
* comparing what the list enumerated against what was asked for. Rule 5.
*/
truncated: boolean
/**
* Messages the batch asked for and did not get back. Not the same as
* `truncated`: truncated means we stopped asking, this means we asked and
* Gmail did not answer. Both render, differently.
*/
failedFetches: number
/**
* What Gmail said about the parts it refused, deduplicated, first few only.
*
* A count is not enough: the same number of failures on every run is not a
* rate limit but the same messages failing each time, for a reason the count
* cannot express. Rule 1: report what happened rather than pick a cause, and
* make the next run able to answer it.
*/
failureSample: string[]
}
/**
* Split a From header into address and display name.
*
* Handles `Name <a@b.com>`, a bare `a@b.com`, and a quoted name containing a
* comma (`"Labs, Upscaler" <news@upscaler.vendor.example>`), which a split on
* ',' or '<' alone gets wrong.
*/
export function parseFrom(header: string): { sender: string; senderName: string } {
const m = /<([^>]+)>\s*$/.exec(header.trim())
const rawAddr = (m ? m[1]! : header).trim()
const name = m ? header.slice(0, m.index).trim().replace(/^"(.*)"$/, '$1').trim() : ''
return { sender: normalizeAddress(rawAddr), senderName: name }
}
/**
* Lowercase, and drop a `+tag`.
*
* Stripping the tag is load-bearing for grouping, not just tidiness. Stripe
* sends failed-payment notices from
* `failed-payments+acct_1ExampleAcct0000@stripe.com`, a per-account tag. Left
* whole, every account becomes its own "sender" in the list and the filter
* button applies to one of them. Normalized, the vendor is one row.
*/
export function normalizeAddress(addr: string): string {
const a = addr.trim().toLowerCase().replace(/^<|>$/g, '')
const at = a.lastIndexOf('@')
if (at < 1) return a
const local = a.slice(0, at).split('+')[0]!
return `${local}${a.slice(at)}`
}
/**
* Split an address-list header (To, Cc) into its entries.
*
* A split on ',' is wrong: a client thread Cc'd to
* `"Ortiz, Lena" <Lena.Ortiz@lakeside-expo.example>` has a display name that
* contains the separator. Splitting on commas turns one recipient into
* `"Ortiz` and `Lena" <Lena.Ortiz@lakeside-expo.example>`: the first is not an
* address at all and the second would be sent to.
*
* So this tracks quoting and angle brackets and splits only on a comma outside
* both. Entries are returned verbatim (name and all); `parseFrom` is what
* reduces one to an address.
*/
export function splitAddressList(header: string): string[] {
const out: string[] = []
let cur = ''
let inQuote = false
let inAngle = false
for (const ch of header) {
if (ch === '"' && !inAngle) inQuote = !inQuote
else if (ch === '<' && !inQuote) inAngle = true
else if (ch === '>' && !inQuote) inAngle = false
if (ch === ',' && !inQuote && !inAngle) {
if (cur.trim()) out.push(cur.trim())
cur = ''
continue
}
cur += ch
}
if (cur.trim()) out.push(cur.trim())
return out
}
/**
* Clean Gmail's snippet for a list row.
*
* Snippets arrive HTML-entity-encoded (' &) and, in marketing mail,
* padded with hundreds of zero-width/soft-hyphen characters to game preview
* text: the "͏ ͏ ͏" seen in every promotional row. Same junk class as
* htmlToText strips from bodies.
*/
export function cleanSnippet(raw: string | undefined, cap = 140): string {
if (!raw) return ''
const s = raw
.replace(/ | /gi, ' ')
.replace(/&/gi, '&')
.replace(/</gi, '<')
.replace(/>/gi, '>')
.replace(/"/gi, '"')
.replace(/'|'/gi, "'")
.replace(/[-‏⁠]/g, '')
.replace(/\s+/g, ' ')
.trim()
return s.length > cap ? s.slice(0, cap - 1).trimEnd() + '…' : s
}
const API = 'https://gmail.googleapis.com/gmail/v1/users/me/messages'
const BATCH_API = 'https://gmail.googleapis.com/batch/gmail/v1'
/**
* Gmail's batch endpoint: many messages.get in one HTTP request.
*
* A batch is one fetch, so a chunk of messages costs one Cloudflare subrequest
* instead of one each, and one round trip instead of many serial ones, which
* is also where the wall-clock goes. Gmail's own per-message quota is
* unchanged, which is why the chunk size below matters.
*
* Partial failure is counted, not swallowed. One unreadable message does not
* fail the mailbox, but silently returning 97 of 100 would make a truncated
* read look like a complete one. The count comes back and the surface says so
* (rule 5).
*
* Messages per batch, and why it is not 100: at Gmail's documented maximum of
* 100, roughly a third of each large mailbox comes back refused, with every id
* attempted and the losses inside the batch.
*
* The arithmetic explains it exactly. `messages.get` costs 5 quota units and
* Gmail allows 250 units per second per user; a chunk of 100 spends 500 at once,
* twice the ceiling, and the overflow is refused. The documented maximum is a
* limit on the request, not permission to spend the quota.
*
* 40 per chunk is 200 units (inside the ceiling with room for the token
* refresh), and BATCH_PACE_MS keeps successive chunks from stacking. Each
* mailbox is a different Gmail user with its own budget, so the seven still run
* in parallel.
*/
const BATCH_CHUNK = 40
/** Spacing between chunks, so 200 units do not land twice in one second. */
const BATCH_PACE_MS = 1_100
/** How many times to re-ask for parts Gmail refused. */
const BATCH_RETRIES = 2
/**
* How long to wait before re-asking, in order.
*
* Sized to the quota, not doubled from the pacing interval. The refusals say,
* in Gmail's own words:
*
* 403 PERMISSION_DENIED: Quota exceeded for quota metric 'Total Query Cost'
* and limit 'Units per MINUTE per user'
*
* A per-minute budget is not repaired by waiting a few seconds; retries inside
* the same exhausted minute are refused for the same reason. These wait long
* enough for the window to actually roll. The sync runs every
* fifteen minutes, so a minute of waiting costs nothing that matters.
*/
const RETRY_BACKOFF_MS = [20_000, 60_000]
/**
* How many list pages to walk before giving up on counting the window.
*
* 500 ids a page, so this covers 10,000 messages for 20 subrequests. Past that
* the count comes back null ("we did not finish counting") rather than a
* number that is really "the first ten thousand".
*/
const LIST_PAGE_CAP = 20
/** One message as Gmail returns it at format=metadata. */
interface RawMessage {
id: string
threadId: string
internalDate?: string
labelIds?: string[]
snippet?: string
payload?: { headers?: { name: string; value: string }[] }
}
/**
* Build the multipart/mixed body. Each part is a complete HTTP request.
*
* The trailing blank line after the request line is required; a part without
* it is rejected as malformed, with an error that names nothing useful.
*/
export function buildBatchBody(ids: string[], boundary: string): string {
const parts = ids.map((id, i) => {
const path =
`/gmail/v1/users/me/messages/${encodeURIComponent(id)}?format=metadata` +
`&metadataHeaders=From&metadataHeaders=Subject&metadataHeaders=Date`
return (
`--${boundary}\r\n` +
`Content-Type: application/http\r\n` +
`Content-ID: <item-${i}>\r\n\r\n` +
`GET ${path}\r\n\r\n`
)
})
return `${parts.join('')}--${boundary}--\r\n`
}
/**
* Pull the message objects out of a multipart/mixed batch response.
*
* Responses are matched by the id in their own body, not by Content-ID order.
* Google documents that parts may come back in any order, and every successful
* body already carries the message's own id, so the mapping is free and cannot
* drift. A part with no parseable message is counted as a failure rather than
* dropped, which is the difference between "this mailbox has 97 messages" and
* "we asked for 100 and got 97".
*/
export function parseBatchResponse(
contentType: string,
body: string,
): { messages: RawMessage[]; failed: number; errors: string[] } {
const m = /boundary=(?:"([^"]+)"|([^;\s]+))/i.exec(contentType)
const boundary = m?.[1] ?? m?.[2]
// No boundary means this is not a multipart response at all: an error page,
// or a shape change. Every part is a failure; none is invented.
if (!boundary) return { messages: [], failed: 0, errors: ['response was not multipart'] }
const messages: RawMessage[] = []
const errors: string[] = []
let failed = 0
for (const chunk of body.split(`--${boundary}`)) {
const trimmed = chunk.trim()
if (!trimmed || trimmed === '--') continue
// Each part is an HTTP response: status line, headers, blank line, JSON.
const at = trimmed.indexOf('{')
if (at === -1) {
failed++
// Keep the status line. A part with no JSON at all still says what
// happened on its first line, and that line is the difference between
// "rate limited" and "these messages are gone": two failures with
// identical counts and opposite fixes.
errors.push(trimmed.split('\r\n').find((l) => l.startsWith('HTTP/')) ?? 'no body')
continue
}
try {
const parsed = JSON.parse(trimmed.slice(at)) as RawMessage & {
error?: { message?: string; status?: string; code?: number }
}
if (parsed.error || typeof parsed.id !== 'string') {
failed++
errors.push(
parsed.error
? `${parsed.error.code ?? '?'} ${parsed.error.status ?? ''} ${parsed.error.message ?? ''}`.trim()
: 'a 200 with no message id',
)
} else {
messages.push(parsed)
}
} catch (err) {
failed++
errors.push(`unparseable part: ${errorMessage(err)}`)
}
}
return { messages, failed, errors }
}
/** A batch boundary that cannot collide with the bodies it separates. */
const batchBoundary = (): string => {
const b = new Uint8Array(12)
crypto.getRandomValues(b)
return `batch-${[...b].map((x) => x.toString(16).padStart(2, '0')).join('')}`
}
/** One pass: every id once, paced. Returns what came back and what did not. */
async function batchPass(
ids: string[],
accessToken: string,
f: typeof fetch,
paceMs: number,
): Promise<{ messages: RawMessage[]; missing: string[]; errors: string[] }> {
const out: RawMessage[] = []
const errors: string[] = []
const got = new Set<string>()
for (let i = 0; i < ids.length; i += BATCH_CHUNK) {
if (i > 0 && paceMs > 0) await sleep(paceMs)
const chunk = ids.slice(i, i + BATCH_CHUNK)
const boundary = batchBoundary()
const res = await f(BATCH_API, {
method: 'POST',
headers: {
authorization: `Bearer ${accessToken}`,
'content-type': `multipart/mixed; boundary=${boundary}`,
},
body: buildBatchBody(chunk, boundary),
})
// A whole chunk refused is not a failed mailbox (the others may be fine),
// and every id in it stays on the missing list for the retry.
if (!res.ok) {
errors.push(`whole chunk refused: HTTP ${res.status}`)
continue
}
const parsed = parseBatchResponse(res.headers.get('content-type') ?? '', await res.text())
errors.push(...parsed.errors)
for (const m of parsed.messages) {
if (!got.has(m.id)) {
got.add(m.id)
out.push(m)
}
}
}
return { messages: out, missing: ids.filter((id) => !got.has(id)), errors }
}
/**
* Every message, with the refused ones asked for again.
*
* Retries exist because the failures are rate limits, not bad messages. A
* refused part means Gmail was busy, and asking again a moment later works,
* which is exactly the case where giving up would quietly shrink a mailbox.
* What remains after the retries is reported as a count, never swallowed.
*/
async function batchGet(
ids: string[],
accessToken: string,
f: typeof fetch,
paceMs: number,
): Promise<{ messages: RawMessage[]; failed: number; lastErrors: string[] }> {
const all: RawMessage[] = []
let pending = ids
let lastErrors: string[] = []
for (let attempt = 0; attempt <= BATCH_RETRIES && pending.length; attempt++) {
if (attempt > 0 && paceMs > 0) await sleep(RETRY_BACKOFF_MS[attempt - 1] ?? 60_000)
const { messages, missing, errors } = await batchPass(pending, accessToken, f, paceMs)
all.push(...messages)
pending = missing
lastErrors = errors
}
return { messages: all, failed: pending.length, lastErrors }
}
/**
* Read one mailbox: `list` for the ids (paged to the end, to count the
* window), then the batch endpoint for the metadata of the newest `maxMessages`.
*
* Never throws, for a caller looping over mailboxes. A dead token on one mailbox
* is routine (it is exactly what happens the day a role account changes hands),
* and losing the other six because of it would be the worse outcome. The failure
* comes back as `ok: false` with its reason, and the surface is required to say
* so rather than render an inbox with nothing in it.
*/
export async function readMailbox(
mailbox: string,
sa: GmailCredential,
opts: {
maxMessages?: number
query?: string
fetchImpl?: typeof fetch
/**
* Milliseconds between batch chunks. Defaults to the real pacing; tests
* pass 0 so a suite that must stay offline and fast is not also waiting out
* a rate limiter it is not exercising.
*/
paceMs?: number
/**
* Walk the id list to the end to learn the window size, or stop as soon as
* `maxMessages` ids are in hand.
*
* True for the sync, which wants the count and can afford up to twenty list
* pages. False for a live search, which runs on the request path with a
* person waiting: twenty pages × seven mailboxes is 140 subrequests and
* several seconds spent counting results nobody asked to count. A search
* says "the first N matches", not "N of M".
*/
countWindow?: boolean
} = {},
): Promise<MailboxRead> {
const f = opts.fetchImpl ?? fetch
const max = opts.maxMessages ?? 100
const paceMs = opts.paceMs ?? BATCH_PACE_MS
const countWindow = opts.countWindow ?? true
// Inbox only, recent only. Archived mail is not a queue anyone is looking at.
const query = opts.query ?? 'in:inbox newer_than:30d'
try {
const { accessToken, scopes } = await accessTokenForCredential(mailbox, sa, f)
const auth = { authorization: `Bearer ${accessToken}` }
// The list is paged to the end; the fetch is not. They cost very different
// things: `messages.list` returns ids only, one subrequest per 500, so
// walking the whole window is nearly free, and it is the only way to learn
// how big the window actually is (see `windowTotal`).
const ids: string[] = []
let pageToken: string | undefined
let pages = 0
do {
const params = new URLSearchParams({ maxResults: '500', q: query })
if (pageToken) params.set('pageToken', pageToken)
const listRes = await f(`${API}?${params.toString()}`, { headers: auth })
if (!listRes.ok) throw new Error(`list failed: HTTP ${listRes.status}`)
const list = (await listRes.json()) as {
messages?: { id: string }[]
nextPageToken?: string
}
for (const m of list.messages ?? []) ids.push(m.id)
pageToken = list.nextPageToken
pages++
} while (pageToken && pages < LIST_PAGE_CAP && (countWindow || ids.length < max))
// Null rather than a wrong number: if the walk stopped at the page cap we
// do not know the total, and saying so is the whole point of this field.
// A leftover pageToken is the only thing that matters here: it means the
// walk stopped early, whether at the page cap or because a search asked not
// to count. Null then: "we did not finish counting" is a different fact
// from any total. When the list genuinely ended, the count is exact and
// free even for a search that never asked for it.
const windowTotal = pageToken ? null : ids.length
// Gmail returns the list newest-first, so the cap keeps the newest.
const wanted = ids.slice(0, max)
const truncated = ids.length > wanted.length || pageToken !== undefined
// One fetch per chunk of messages, not one per message. See BATCH_CHUNK.
const { messages: raw, failed, lastErrors } = await batchGet(wanted, accessToken, f, paceMs)
const messages: InboxMessage[] = raw.map((m) => {
const header = (n: string): string =>
m.payload?.headers?.find((h) => h.name.toLowerCase() === n)?.value ?? ''
const { sender, senderName } = parseFrom(header('from'))
return {
id: m.id,
threadId: m.threadId,
mailbox,
sender,
senderName,
subject: header('subject'),
snippet: cleanSnippet(m.snippet),
// internalDate over the Date header: the header is set by the sender and
// is routinely wrong or absent, and this list sorts by it.
date: m.internalDate
? new Date(Number(m.internalDate)).toISOString()
: new Date(0).toISOString(),
labels: m.labelIds ?? [],
unread: (m.labelIds ?? []).includes('UNREAD'),
}
})
// Gmail returns batch parts in no guaranteed order, so sort rather than
// trusting arrival. The page sorts again, but a list that is only sorted
// downstream is one bad refactor away from arriving shuffled.
messages.sort((a, b) => (a.date < b.date ? 1 : a.date > b.date ? -1 : 0))
return {
mailbox,
ok: true,
error: null,
messages,
canSend: scopes === null ? null : scopes.includes(GMAIL_SEND_SCOPE),
windowTotal,
truncated,
failedFetches: failed,
failureSample: [...new Set(lastErrors)].slice(0, 3),
}
} catch (err) {
const reason = errorMessage(err)
// canSend stays null on a failed read: we never got far enough to learn
// anything about the token's scopes, and "failed" must not be reported as
// "read-only"; they need different fixes.
return {
mailbox,
ok: false,
error: reason,
messages: [],
canSend: null,
windowTotal: null,
truncated: false,
failedFetches: 0,
failureSample: [],
}
}
}
/**
* One message, in full, right now.
*
* Called from the request path rather than the cron, so it costs a token
* refresh plus one Gmail call per open: about a fifth of a second, and only for
* the message someone actually clicked.
*
* Returns `{ error }` when the id is unknown or unreadable. The caller renders
* that as "could not be read" and never as an empty message: a body we failed to fetch
* and a body that is genuinely blank are different facts.
*/
export async function fetchBody(
mailbox: string,
id: string,
sa: GmailCredential,
fetchImpl: typeof fetch = fetch,
): Promise<MessageBody | { error: string }> {
try {
const { accessToken } = await accessTokenForCredential(mailbox, sa, fetchImpl)
const res = await fetchImpl(`${API}/${encodeURIComponent(id)}?format=full`, {
headers: { authorization: `Bearer ${accessToken}` },
})
if (!res.ok) return { error: `Gmail returned HTTP ${res.status} for that message.` }
const m = (await res.json()) as {
id: string
threadId?: string
internalDate?: string
labelIds?: string[]
payload?: Parameters<typeof extractBody>[0] & { headers?: { name: string; value: string }[] }
}
const header = (n: string): string =>
m.payload?.headers?.find((h) => h.name.toLowerCase() === n)?.value ?? ''
const { sender, senderName } = parseFrom(header('from'))
const { text, kind, html } = extractBody(m.payload)
// Attachments: name, mime type, size and Gmail's handle, enough for the
// page to say how big a file is and for /inbox/attachment to fetch it. See
// the Attachment interface for why an index rather than an id addresses a
// download.
const attachments = collectAttachments(m.payload)
return {
id: m.id,
mailbox,
sender,
senderName,
subject: header('subject'),
to: header('to'),
date: m.internalDate ? new Date(Number(m.internalDate)).toISOString() : new Date(0).toISOString(),
text: text.slice(0, BODY_CAP),
kind,
truncated: text.length > BODY_CAP,
html: html ? html.slice(0, HTML_CAP) : null,
htmlTruncated: html.length > HTML_CAP,
unread: (m.labelIds ?? []).includes('UNREAD'),
attachments,
// Threading and reply-addressing. format=full already returns every
// header, so these cost nothing extra, and they are read here, on the
// server, because the reply's recipient must not come from a form field.
// An absent header is null (or '' for the two string fields), never
// undefined: the caller branches on it and `undefined` would read as a
// bug rather than as "the sender did not send one".
threadId: m.threadId ?? '',
messageId: header('message-id') || null,
references: header('references') || null,
replyTo: header('reply-to'),
cc: header('cc'),
}
} catch (err) {
return { error: errorMessage(err) }
}
}
/**
* How big an attachment this board will hand back.
*
* A memory limit, not a policy one, which is why it is larger than the send
* cap in config/inbox.ts. What a client attaches is not our choice; what the
* Worker can hold is. The chain for one download: Gmail answers with base64
* (4/3 of the bytes) inside a JSON string, which V8 holds as UTF-16 (×2), and
* the decoded bytes exist alongside it, so a 10 MB file peaks near 40 MB
* against a Worker's 128 MB. Above this the page says the size and sends the
* reader to Gmail, which is a true answer; a truncated file would not be.
*/
export const ATTACHMENT_DOWNLOAD_CAP = 10 * 1024 * 1024
/**
* One attachment's bytes.
*
* Never throws, and returns the size it actually got rather than trusting the
* size the message listing reported: the caller checks the cap before calling
* (from the part's declared size) and this returns what arrived, so a message
* that under-declares a part cannot walk past the ceiling unnoticed.
*/
export async function fetchAttachment(
mailbox: string,
messageId: string,
attachmentId: string,
sa: GmailCredential,
fetchImpl: typeof fetch = fetch,
): Promise<{ bytes: Uint8Array } | { error: string }> {
try {
const { accessToken } = await accessTokenForCredential(mailbox, sa, fetchImpl)
const res = await fetchImpl(
`${API}/${encodeURIComponent(messageId)}/attachments/${encodeURIComponent(attachmentId)}`,
{ headers: { authorization: `Bearer ${accessToken}` } },
)
if (!res.ok) return { error: `Gmail returned HTTP ${res.status} for that attachment.` }
const body = (await res.json()) as { data?: string; size?: number }
if (!body.data) return { error: 'Gmail returned that attachment with no data.' }
const bytes = decodeB64UrlBytes(body.data)
if (bytes.byteLength > ATTACHMENT_DOWNLOAD_CAP) {
return {
error: `That attachment is ${bytes.byteLength.toLocaleString()} bytes, over this board's ${ATTACHMENT_DOWNLOAD_CAP.toLocaleString()}-byte download ceiling. It is in Gmail.`,
}
}
return { bytes }
} catch (err) {
return {
error: errorMessage(err),
}
}
}