server/providers.ts
import { Readable } from "node:stream";
import { ATTACHMENT_BYTES, ATTACHMENT_COUNT, MIME_BYTES, attachmentSize } from "../shared/attachments.js";
import { AttachmentPreflightError, attachmentWork, mimeAttachments, storeAttachmentBytes, gmailJson } from "./attachment-bytes.js";
import { COMPOSE_HTML_LIMIT, validComposeMarkup } from "../shared/compose-content.js";
import { parseUnsubscribe } from "../shared/unsubscribe.js";
import { parseICalendar, generateICalendar } from "./ical.js";
import { google } from "./google-api.js";
import { createHash } from "node:crypto";
import { isDeepStrictEqual } from "node:util";
import { ImapFlow, type SearchObject, type FetchMessageObject, type MessageStructureObject } from "imapflow";
import type { gmail_v1 } from "./google-api.js";
import { simpleParser } from "mailparser";
import { workLimit } from "./resource-limits.js";
import { canonicalBase64Size, checkMimeSource, decodeGmailSource, gmailResponseLimit, imapSourceQuery, parseMailSource, rawMailLimit } from "./mime-limits.js";
import { Parser } from "htmlparser2";
import { imageSource } from "../shared/types.js";
import { boundedMailHtml, previewText } from "../shared/mail-content.js";
import nodemailer from "nodemailer";
import MailComposer from "nodemailer/lib/mail-composer/index.js";
import { z } from "zod";
import type { GmailLabel, Mail, MailImage, MailAttachment, MailSummary, Provider, ProviderSearchResult, SyncFailure } from "../shared/types.js";
import type { ProviderSearchPlan } from "../shared/mail-search.js";
import type { StoredAccount } from "./store.js";
export class SyncError extends Error {
constructor(public failure: SyncFailure, provider: Provider) {
const messages: Record<SyncFailure["kind"], string> = {
auth: provider === "gmail"
? "Google sign-in was rejected. Continue with Google in Settings."
: "Authentication was rejected. Replace the app password in Settings.",
access: "Mailbox access was denied. Check provider permissions or administrator policy.",
transient: "The provider or connection is temporarily unavailable. Retry connection later.",
message: "An email could not be read. Retry connection to fetch it again.",
unknown: "Sync failed. Retry connection; if it persists, check the safe failure details.",
};
super(messages[failure.kind]);
}
}
export function syncError(error: unknown, provider: Provider, phase: SyncFailure["phase"]): SyncError {
if (error instanceof SyncError) return error;
const e = error as {
code?: unknown; name?: unknown; status?: unknown; authenticationFailed?: boolean; serverResponseCode?: unknown;
response?: { status?: unknown; data?: { error?: string | { errors?: { reason?: unknown }[] } } };
} | null;
const statusValue = e?.response?.status ?? e?.status ?? e?.code;
const status = typeof statusValue === "number" && statusValue >= 400 && statusValue <= 599 ? statusValue : undefined;
const detail = e?.response?.data?.error;
const reasons = typeof detail === "object" && Array.isArray(detail?.errors) ? detail.errors.map((item) => item?.reason) : [];
const rateLimit = status === 429 || reasons.some((reason) => ["rateLimitExceeded", "userRateLimitExceeded", "quotaExceeded", "dailyLimitExceeded"].includes(typeof reason === "string" ? reason : ""));
const codes = ["invalid_grant", "invalid_token", "EAUTH", "AUTHENTICATIONFAILED", "ETHROTTLE", "ETIMEDOUT", "ETIMEOUT", "ECONNRESET", "ECONNREFUSED", "ECONNABORTED", "EPIPE", "ENOTFOUND", "EAI_AGAIN", "INCOMPLETE_LIST", "CONNECTION_CLOSED", "READ_FAILED", "PARSE_FAILED", "CLEANUP_FAILED"];
const candidate = typeof detail === "string" ? detail : e?.code ?? e?.serverResponseCode;
let code = typeof candidate === "string" && codes.includes(candidate) ? candidate : "UNKNOWN";
if ([e?.name, candidate].some((value) => value === "AbortError" || value === "TimeoutError") || candidate === "CONNECT_TIMEOUT" || candidate === "GREETING_TIMEOUT") code = "ETIMEDOUT";
if (candidate === "NoConnection" || candidate === "EConnectionClosed") code = "CONNECTION_CLOSED";
const auth = e?.authenticationFailed || status === 401 || ["invalid_grant", "invalid_token", "EAUTH", "AUTHENTICATIONFAILED"].includes(code);
const transient = rateLimit || (status !== undefined && status >= 500) || ["ETHROTTLE", "ETIMEDOUT", "ETIMEOUT", "ECONNRESET", "ECONNREFUSED", "ECONNABORTED", "EPIPE", "ENOTFOUND", "EAI_AGAIN", "INCOMPLETE_LIST", "CONNECTION_CLOSED"].includes(code);
const kind = auth ? "auth" : transient ? "transient" : status === 403 ? "access" : phase === "read" || phase === "parse" ? "message" : "unknown";
if (code === "UNKNOWN") code = auth ? "AUTH_REJECTED" : rateLimit ? "RATE_LIMIT" : status === 403 ? "ACCESS_DENIED" : status ? "HTTP_ERROR" : phase === "parse" ? "PARSE_FAILED" : phase === "read" ? "READ_FAILED" : "UNKNOWN";
return new SyncError({ kind, phase: code === "invalid_grant" || code === "invalid_token" ? "auth" : phase, code, ...(status ? { status } : {}) }, provider);
}
async function syncOperation<T>(provider: Provider, phase: SyncFailure["phase"], work: () => Promise<T>): Promise<T> {
try { return await work(); }
catch (error) { throw syncError(error, provider, phase); }
}
const imapFailures = new WeakMap<ImapFlow, { failed: Promise<never>; error?: SyncError }>();
export async function imapOperation<T>(client: ImapFlow, work: () => Promise<T>): Promise<T> {
const state = imapFailures.get(client)!;
if (state.error) throw state.error;
const result = await Promise.race([work(), state.failed]);
if (state.error) throw state.error;
return result;
}
async function* imapFetch(client: ImapFlow, ...args: Parameters<ImapFlow["fetch"]>) {
const iterator = client.fetch(...args);
try {
while (true) {
const next = await imapOperation(client, () => iterator.next());
if (next.done) return;
yield next.value;
}
} finally {
// Closing the transport releases ImapFlow's command; return drains iterator backpressure.
void iterator.return?.().catch(() => {});
}
}
export const presets = {
"gmail-imap": {
imapHost: "imap.gmail.com",
smtpHost: "smtp.gmail.com",
imapPort: 993,
smtpPort: 465,
},
icloud: {
imapHost: "imap.mail.me.com",
smtpHost: "smtp.mail.me.com",
imapPort: 993,
smtpPort: 587,
},
fastmail: {
imapHost: "imap.fastmail.com",
smtpHost: "smtp.fastmail.com",
imapPort: 993,
smtpPort: 465,
},
yahoo: {
imapHost: "imap.mail.yahoo.com",
smtpHost: "smtp.mail.yahoo.com",
imapPort: 993,
smtpPort: 465,
},
};
export function oauth(origin?: string) {
return new google.auth.OAuth2(
process.env.GOOGLE_CLIENT_ID,
process.env.GOOGLE_CLIENT_SECRET,
`${origin || process.env.APP_URL || "http://127.0.0.1:5173"}/api/oauth/google/callback`,
);
}
export function gmail(a: StoredAccount, timeout = 30000) {
const auth = oauth();
auth.setCredentials({ refresh_token: a.credentials.refreshToken });
auth.transporter.defaults.timeout = timeout;
auth.transporter.defaults.retry = false;
auth.transporter.defaults.retryConfig = { retry: 0 };
return google.gmail({ version: "v1", auth });
}
export async function gmailHistoryId(a: StoredAccount): Promise<string> {
const result = await syncOperation(a.provider, "list", () => gmail(a, 10000).users.getProfile(
{ userId: "me", fields: "historyId" }, { timeout: 10000, retry: false },
));
const id = result.data.historyId;
if (typeof id !== "string" || !/^\d+$/.test(id)) throw new Error("Gmail did not return a mailbox revision.");
return id;
}
export async function listGmailLabels(a: StoredAccount): Promise<GmailLabel[]> {
const result = await syncOperation(a.provider, "list", () => gmail(a).users.labels.list({ userId: "me" }, { timeout: 30000, retry: false }));
return (result.data.labels || []).flatMap((label) =>
label.id && label.name && (label.type === "user" || label.type === "system")
? [{ id: label.id, accountId: a.id, name: label.name, type: label.type }]
: [],
);
}
export async function updateGmailLabels(a: StoredAccount, m: Mail, labelIds: string[], customIds: string[]) {
const api = gmail(a);
const current = await api.users.messages.get({
userId: "me", id: m.remoteId, format: "metadata", fields: "labelIds",
});
const removeLabelIds = (current.data.labelIds || []).filter((id) => customIds.includes(id) && !labelIds.includes(id));
if (removeLabelIds.length > 100) throw new Error("Too many labels to remove at once. Update this message in Gmail.");
const result = await api.users.messages.modify({
userId: "me",
id: m.remoteId,
requestBody: {
addLabelIds: labelIds,
// Remove only this account's custom labels, including labels absent from an older cache.
removeLabelIds,
},
});
if (result.data.id !== m.remoteId) throw new Error("Gmail did not confirm the labels. Sync and try again.");
return result.data.labelIds || [];
}
export function imap(a: StoredAccount, watch = false) {
const client = new ImapFlow({
host: a.credentials.imapHost!,
port: a.credentials.imapPort || 993,
secure: true,
auth: {
user: a.credentials.username || a.email,
pass: a.credentials.password!,
},
logger: false,
maxLineLength: 65536,
maxLiteralSize: rawMailLimit + 1,
maxResponseSize: rawMailLimit + 65536,
connectionTimeout: 15000,
greetingTimeout: 15000,
socketTimeout: watch ? 30 * 60000 : 30000,
...(watch ? { maxIdleTime: 5 * 60000 } : {}),
});
let reject!: (error: SyncError) => void;
const failed = new Promise<never>((_resolve, fail) => { reject = fail; });
void failed.catch(() => {});
const state: { failed: Promise<never>; error?: SyncError } = { failed };
imapFailures.set(client, state);
client.on("error", (error) => {
state.error ||= syncError(error, a.provider, "connect");
reject(state.error);
client.close();
});
return client;
}
export type Incoming = Omit<
Mail,
"category" | "relevance" | "reason" | "archived" | "snoozedUntil"
>;
const parseLimit = workLimit(2, 16);
export function parse(...args: Parameters<typeof parseMessage>): Promise<Incoming> {
return parseLimit(() => parseMessage(...args));
}
async function parseMessage(
a: StoredAccount,
remoteId: string,
source: Buffer,
unread: boolean,
date?: number,
): Promise<Incoming> {
checkMimeSource(source);
const m = await parseMailSource(source);
const decodedBytes = Buffer.byteLength(m.text || "") + Buffer.byteLength(m.html || "") + m.attachments.reduce((size, part) => size + part.content.length, 0);
if (decodedBytes > 16 * 1024 * 1024 || m.attachments.length > 128 || m.headerLines.length > 512)
throw new Error("Decoded email limit exceeded.");
const from = m.from?.value[0];
const body = m.text || "";
let total = 0;
let imported = 0;
const attachments: MailAttachment[] = [];
for (const [index, attachment] of m.attachments.entries()) {
const filename = attachment.filename || "attachment";
const contentType = attachment.contentType || "application/octet-stream";
const id = createHash("sha256").update(`${a.id}:${remoteId}:${index}:`).update(attachment.content).digest("hex");
const eligible = imported < 10 && attachment.content.length <= 524288 && total + attachment.content.length <= 1048576;
let valid = eligible && attachmentMetadataSchema.safeParse({ filename, contentType }).success;
const content = valid ? attachment.content.toString("base64") : "";
if (valid) {
const { get, put } = await import("./store.js");
const current = get("account", a.id);
valid = Boolean(current && current.id === a.id && current.provider === a.provider && current.email === a.email && isDeepStrictEqual(current.credentials, a.credentials));
if (valid) {
put("attachment", `${a.id}:${remoteId}:${id}`, { accountId: a.id, content });
total += attachment.content.length; imported++;
}
}
const contentId = attachment.contentId?.replace(/^<|>$/g, "").trim();
attachments.push({ id, sourceIndex: index, contentId: contentId && contentId.length <= 500 ? contentId : null, filename: filename.replace(/[\u0000-\u001f\u007f/\\]/g, "_").slice(0, 255), contentType: /^[\w!#$&^.+-]+\/[\w!#$&^.+-]+$/.test(contentType) ? contentType : "application/octet-stream", size: attachment.size, ...(!valid ? { unavailable: true } : {}) });
}
const calendars = m.attachments.filter((part) => part.contentType.toLowerCase() === "text/calendar");
const invitation = calendars.length > 1 ? { status: "unsupported" as const, reason: "multiple-calendars" } : calendars.length ? calendars[0].content.length > 131072 ? { status: "invalid" as const, reason: "limit" } : parseICalendar(calendars[0].content.toString("utf8")) : undefined;
return {
invitation,
unsubscribe: parseUnsubscribe(m.headerLines),
id: `${a.id}:${remoteId}`,
accountId: a.id,
remoteId,
sender: from?.name || from?.address || "Unknown sender",
address: from?.address || "",
to: Array.isArray(m.to)
? m.to.map((x) => x.text).join(", ")
: m.to?.text || a.email,
subject: m.subject || "(No subject)",
preview: previewText(body).slice(0, 220),
body,
images: extractImages(m.html || ""),
html: boundedMailHtml(m.html),
date: date || m.date?.getTime() || Date.now(),
unread,
messageId: m.messageId,
inReplyTo: m.inReplyTo,
attachments,
cc: Array.isArray(m.cc) ? m.cc.map((x) => x.text).join(", ") : m.cc?.text,
replyTo: m.replyTo?.value[0]?.address,
references:
typeof m.references === "string" ? [m.references] : m.references,
};
}
function extractImages(html: string): MailImage[] {
const images: MailImage[] = [];
const seen = new Set<string>();
new Parser({
onopentag(name, attributes) {
if (name !== "img" || images.length >= 50) return;
const src = imageSource(attributes.src || "");
if (!src || seen.has(src)) return;
seen.add(src);
images.push({ src, alt: (attributes.alt || "").slice(0, 500) });
},
}).end(html);
return images;
}
export async function validateImap(a: StoredAccount) {
const client = imap(a);
try {
await client.connect();
await client.mailboxOpen("INBOX");
await smtp(a).verify();
} finally {
if (client.usable) await client.logout();
else client.close();
}
}
export class SentCopyError extends Error {}
// `peek` reads summary metadata only; it lets sync answer flag questions
// without reopening the encrypted record for unchanged mail.
type CachedMail = Pick<ReadonlyMap<string, Mail>, "size" | "get" | "has"> & { peek?: (id: string) => MailSummary | undefined };
// Sync batches may carry a stored summary for unchanged mail instead of its
// full record.
type SyncBatch = Incoming | MailSummary;
// Bound the identifiers retained from each IMAP search response.
const IMAP_SEARCH_RESULTS_MAX = 10000;
// Keep a bounded min-heap so an oversized response does not require another
// full-size array or sorting every returned UID.
function newestSearchUids(found: number[]): number[] {
const heap = found.slice(0, IMAP_SEARCH_RESULTS_MAX);
const siftDown = (start: number) => {
let parent = start;
while (parent * 2 + 1 < heap.length) {
let child = parent * 2 + 1;
if (child + 1 < heap.length && heap[child + 1] < heap[child]) child++;
if (heap[parent] <= heap[child]) break;
[heap[parent], heap[child]] = [heap[child], heap[parent]];
parent = child;
}
};
for (let index = Math.floor(heap.length / 2) - 1; index >= 0; index--) siftDown(index);
for (let index = IMAP_SEARCH_RESULTS_MAX; index < found.length; index++) {
if (found[index] > heap[0]) { heap[0] = found[index]; siftDown(0); }
}
return heap;
}
// `known` receives a cache ID and Message-ID so cached matches are not downloaded again.
export type ProviderSearchOptions = { limit: number; signal: AbortSignal; known: (id: string, messageId?: string) => boolean };
type ProviderSearchFetch = Pick<ProviderSearchResult, "examined" | "complete" | "narrowed" | "mailboxes"> & { messages: Incoming[] };
// Gmail has no body-only operator, so body: searches the whole message and an
// excluded body: term is left to the cached search. Values are quoted so text
// such as OR or label:x stays text instead of becoming a Gmail operator.
export function gmailSearchQuery(plan: ProviderSearchPlan) {
const narrowed: ProviderSearchResult["narrowed"] = [...plan.deviceOnly];
const parts = plan.terms.flatMap((term) => {
if (term.kind === "read") return [term.read ? "is:read" : "is:unread"];
// Epoch seconds avoid Gmail's own day boundaries; after: widens by one second.
if (term.kind === "date") return [term.edge === "after" ? `after:${term.at / 1000 - 1}` : `before:${term.at / 1000}`];
const value = `"${term.value}"`;
if (term.field === "body") {
if (!narrowed.includes("body:")) narrowed.push("body:");
return term.exclude ? [] : [value];
}
if (term.field === "to") return term.exclude ? [`-to:${value}`, `-cc:${value}`] : [`{to:${value} cc:${value}}`];
return [`${term.exclude ? "-" : ""}${term.field === "any" ? "" : `${term.field}:`}${value}`];
});
return { q: parts.join(" "), narrowed };
}
// ImapFlow cannot repeat a key in one query, so terms are packed into groups of
// distinct keys and the caller intersects one UID SEARCH per group. IMAP dates
// use the server's day, so each bound widens by one day for the cached search.
export function imapSearchGroups(plan: ProviderSearchPlan) {
const day = 86400000;
const groups: SearchObject[] = [];
for (const term of plan.terms) {
let key: SearchObject;
if (term.kind === "read") key = { seen: term.read };
else if (term.kind === "date") key = term.edge === "after" ? { since: new Date(term.at - day) } : { before: new Date(term.at + day) };
else {
const match: SearchObject = term.field === "any" ? { text: term.value } : term.field === "to" ? { or: [{ to: term.value }, { cc: term.value }] } : { [term.field]: term.value };
key = term.exclude ? { not: match } : match;
}
const group = groups.find((current) => Object.keys(key).every((name) => !(name in current)));
if (group) Object.assign(group, key);
else groups.push(key);
}
return { groups, narrowed: [...plan.deviceOnly] };
}
// All providers expose the same read, update and send operations to the client.
export const providers: Record<
Provider,
{
sent: (a: StoredAccount, bound: number) => Promise<{ messages: Incoming[]; truncated: boolean }>;
search: (a: StoredAccount, plan: ProviderSearchPlan, options: ProviderSearchOptions) => Promise<ProviderSearchFetch>;
thread?: (a: StoredAccount, threadId: string) => Promise<Incoming[]>;
list: (a: StoredAccount, cached: CachedMail, onBatch?: (batch: SyncBatch[]) => void, onWarning?: (warning: SyncFailure) => void, labels?: GmailLabel[], retain?: boolean) => Promise<Incoming[]>;
images: (a: StoredAccount, m: Mail) => Promise<MailImage[]>;
content: (a: StoredAccount, m: Mail) => Promise<{ html: string | null; images: MailImage[]; attachments?: MailAttachment[]; invitation?: Mail["invitation"] }>;
update: (
a: StoredAccount,
m: Mail,
action: "archive" | "read" | "unread" | "inbox" | "trash",
) => Promise<void | { remoteId?: string; mailbox?: Mail["mailbox"] }>;
send: (a: StoredAccount, input: SendInput) => Promise<void>;
}
> & { gmail: { thread: (a: StoredAccount, threadId: string) => Promise<Incoming[]> } } = {
gmail: {
async sent(a, bound) {
z.number().int().min(1).max(100).parse(bound);
const api = gmail(a);
const messages: Incoming[] = [];
let pageToken: string | undefined;
const pages = new Set<string | undefined>();
do {
if (pages.has(pageToken) || pages.size >= 100) throw new Error("Sent listing exceeded its retrieval bound.");
pages.add(pageToken);
const page = await syncOperation(a.provider, "list", () => api.users.messages.list({ userId: "me", labelIds: ["SENT"], maxResults: bound - messages.length, pageToken }, { timeout: 30000, retry: false }));
if ((page.data.messages?.length || 0) > bound - messages.length) throw new Error("Sent listing exceeded its retrieval bound.");
for (const item of page.data.messages || []) {
if (!item.id || messages.some((m) => m.remoteId === item.id)) throw new Error("Sent email identity is unavailable.");
const result = await syncOperation(a.provider, "read", () => api.users.messages.get({ userId: "me", id: item.id!, format: "raw" }, { timeout: 30000, retry: false, maxContentLength: gmailResponseLimit }));
if (!result.data.raw || (result.data.id && result.data.id !== item.id) || !result.data.labelIds?.includes("SENT")) throw new Error("Sent email source or identity is unavailable.");
messages.push({ ...(await parse(a, item.id, decodeGmailSource(result.data.raw), false, Number(result.data.internalDate))), direction: "sent", threadId: result.data.threadId || item.threadId || undefined, labelIds: result.data.labelIds || [] });
if (messages.length >= bound) break;
}
pageToken = page.data.nextPageToken || undefined;
} while (pageToken && messages.length < bound);
return { messages, truncated: !!pageToken };
},
async search(a, plan, { limit, signal, known }) {
const { q, narrowed } = gmailSearchQuery(plan);
const api = gmail(a);
const options = { timeout: 30000, retry: false, signal };
const ids: string[] = [];
const pages = new Set<string | undefined>();
let pageToken: string | undefined;
do {
if (pages.has(pageToken) || pages.size >= limit) throw new Error("Gmail search exceeded its retrieval bound.");
pages.add(pageToken);
signal.throwIfAborted();
const page = await syncOperation(a.provider, "list", () => api.users.messages.list({ userId: "me", q, maxResults: limit - ids.length, pageToken }, options));
const items = page.data.messages || [];
if (items.length > limit - ids.length) throw new Error("Gmail search exceeded its retrieval bound.");
for (const item of items) {
if (!item.id || ids.includes(item.id)) throw new Error("Gmail search returned an unavailable email identity.");
ids.push(item.id);
}
pageToken = page.data.nextPageToken || undefined;
} while (pageToken && ids.length < limit);
const missing = ids.filter((id) => !known(`${a.id}:${id}`));
const messages: Incoming[] = [];
// Read two at a time, as sync does, checking cancellation before each request.
for (let offset = 0; offset < missing.length; offset += 2)
messages.push(...await Promise.all(missing.slice(offset, offset + 2).map(async (id) => {
signal.throwIfAborted();
const result = await syncOperation(a.provider, "read", () => api.users.messages.get({ userId: "me", id, format: "raw" }, { ...options, maxContentLength: gmailResponseLimit }));
if (result.data.id && result.data.id !== id) throw new Error("Gmail search returned an unavailable email identity.");
const labelIds = result.data.labelIds || [];
const parsed = await syncOperation(a.provider, "parse", () => parse(a, id, decodeGmailSource(result.data.raw), labelIds.includes("UNREAD"), Number(result.data.internalDate)));
return { ...parsed, threadId: result.data.threadId || undefined, labelIds, direction: labelIds.includes("SENT") && !labelIds.includes("INBOX") ? "sent" as const : "received" as const };
})));
return { messages, examined: ids.length, complete: !pageToken, narrowed };
},
async thread(a, threadId) {
z.string().min(1).max(100).regex(/^[\w-]+$/).parse(threadId);
const api = gmail(a);
const result = await syncOperation(a.provider, "read", () => api.users.threads.get({ userId: "me", id: threadId, format: "minimal" }, { timeout: 30000, retry: false }));
const items = result.data.messages || [];
if (items.length > 100) throw new Error("This conversation exceeds the 100 message retrieval bound.");
const messages: Incoming[] = [];
for (const item of items) {
if (!item.id) throw new Error("Conversation email identity is unavailable.");
const message = await syncOperation(a.provider, "read", () => api.users.messages.get({ userId: "me", id: item.id!, format: "raw" }, { timeout: 30000, retry: false, maxContentLength: gmailResponseLimit }));
if (!message.data.raw) throw new Error("Conversation email source is unavailable.");
if (message.data.threadId !== threadId) throw new Error("Conversation identity changed. Sync before opening it.");
messages.push({ ...(await parse(a, item.id, decodeGmailSource(message.data.raw), message.data.labelIds?.includes("UNREAD") || false, Number(message.data.internalDate))), direction: message.data.labelIds?.includes("SENT") && !message.data.labelIds?.includes("INBOX") ? "sent" : "received", threadId, labelIds: message.data.labelIds || [] });
}
return messages;
},
async list(a, cached, onBatch, _onWarning, labels, retain = true) {
const auth = oauth();
auth.setCredentials({ refresh_token: a.credentials.refreshToken });
// Bound token refresh as well as message reads, using request cancellation.
auth.transporter.defaults.timeout = 30000;
auth.transporter.defaults.retry = false;
auth.transporter.defaults.retryConfig = { retry: 0 };
const api = google.gmail({ version: "v1", auth });
const options = { timeout: 30000, retry: false };
const messages: Incoming[] = [];
const unreadIds = new Set<string>();
let pageToken: string | undefined;
// Listing unread IDs replaces a metadata request for every cached message.
if (cached.size) {
do {
const page = await syncOperation(a.provider, "list", () => api.users.messages.list({
userId: "me", labelIds: ["INBOX", "UNREAD"], maxResults: 500, pageToken,
}, options));
for (const item of page.data.messages || []) unreadIds.add(item.id!);
pageToken = page.data.nextPageToken || undefined;
} while (pageToken);
}
const memberships = new Map<string, string[]>();
// Refresh labels by membership lists without downloading cached messages.
if (cached.size && labels) {
for (const label of labels.filter((label) => !["INBOX", "UNREAD"].includes(label.id))) {
do {
const page = await syncOperation(a.provider, "list", () => api.users.messages.list({
userId: "me", labelIds: ["INBOX", label.id], maxResults: 500, pageToken,
}, options));
for (const item of page.data.messages || []) {
if (!item.id || !cached.has(`${a.id}:${item.id}`)) continue;
const ids = memberships.get(item.id) || [];
ids.push(label.id);
memberships.set(item.id, ids);
}
pageToken = page.data.nextPageToken || undefined;
} while (pageToken);
}
}
do {
const page = await syncOperation(a.provider, "list", () => api.users.messages.list({
userId: "me",
labelIds: ["INBOX"],
maxResults: 100,
pageToken,
}, options));
const items = page.data.messages || [];
let pageBatch: SyncBatch[] = [];
for (let offset = 0; offset < items.length; offset += 2) {
const results = await Promise.allSettled(
items.slice(offset, offset + 2).map(async (item) => {
const unread = unreadIds.has(item.id!);
const labelIds = labels ? ["INBOX", ...(unread ? ["UNREAD"] : []), ...(memberships.get(item.id!) || [])] : undefined;
const threadId = item.threadId || undefined;
// Summary metadata answers flag and label questions without
// reopening the encrypted record.
const summary = cached.peek?.(`${a.id}:${item.id}`);
if (summary && !retain && onBatch && summary.unread === unread &&
(labelIds === undefined || isDeepStrictEqual(labelIds, summary.labelIds)) &&
(threadId === undefined || threadId === summary.threadId))
return { ...summary, unread };
const previous = cached.get(`${a.id}:${item.id}`);
if (previous) return {
...previous, unread,
...(labelIds !== undefined ? { labelIds } : {}),
...(threadId ? { threadId } : {}),
};
try {
const result = await api.users.messages.get({ userId: "me", id: item.id!, format: "raw" }, { ...options, maxContentLength: gmailResponseLimit });
const unread = result.data.labelIds?.includes("UNREAD") || false;
let source: Buffer;
try { source = decodeGmailSource(result.data.raw); }
catch (error) {
if (error instanceof Error && /byte limit/i.test(error.message)) return syncOperation(a.provider, "parse", () => recoverGmailMetadata(a, item.id!, api));
throw syncError(error, a.provider, "parse");
}
return { ...(await syncOperation(a.provider, "parse", () => parse(a, item.id!, source, unread, Number(result.data.internalDate)))),
threadId: result.data.threadId || undefined, labelIds: result.data.labelIds || [] };
} catch (error) {
if (error instanceof Error && /(?:byte limit|maxContentLength|maximum.*size|response.*size)/i.test(error.message))
return recoverGmailMetadata(a, item.id!, api);
throw syncError(error, a.provider, "read");
}
}),
);
const batch = results.flatMap((result) => result.status === "fulfilled" ? [result.value] : []);
// Retained lists hold full records only; summary rows appear only in
// progressive batch mode.
if (retain || !onBatch) messages.push(...(batch.filter((m) => "body" in m) as Incoming[]));
pageBatch.push(...batch);
const failure = results.find((result) => result.status === "rejected");
// Fresh mail stays progressive; cached-only changes share a page write.
if (failure || batch.some((m) => !cached.has(m.id))) {
onBatch?.(pageBatch);
pageBatch = [];
}
if (failure?.status === "rejected") throw failure.reason;
}
if (pageBatch.length) onBatch?.(pageBatch);
pageToken = page.data.nextPageToken || undefined;
} while (pageToken);
return messages;
},
async images(a, m) {
return (await providers.gmail.content(a, m)).images;
},
async content(a, m) {
const result = await gmail(a).users.messages.get({
userId: "me", id: m.remoteId, format: "raw",
}, { timeout: 30000, retry: false, maxContentLength: gmailResponseLimit });
if (!result.data.raw) throw new Error("Email source is unavailable.");
const parsed = await parse(a, m.remoteId, decodeGmailSource(result.data.raw), m.unread);
return { html: parsed.html ?? null, images: parsed.images!, attachments: parsed.attachments, ...(parsed.invitation ? { invitation: parsed.invitation } : {}) };
},
async update(a, m, action) {
if (action === "trash") {
await gmail(a).users.messages.trash({ userId: "me", id: m.remoteId });
return;
}
await gmail(a).users.messages.modify({
userId: "me",
id: m.remoteId,
requestBody: {
...(action === "unread" || action === "inbox"
? { addLabelIds: [action === "unread" ? "UNREAD" : "INBOX"] }
: { removeLabelIds: [action === "archive" ? "INBOX" : "UNREAD"] }),
},
});
},
async send(a, input) {
input = sendSchema.parse(input);
if (input.attachments?.some(file => "id" in file)) return sendReferenced(a, input);
const message = new MailComposer({
from: a.email,
to: input.to,
cc: input.cc,
bcc: input.bcc,
subject: input.subject,
text: input.body,
html: input.html,
inReplyTo: input.inReplyTo,
references: input.references || input.inReplyTo,
icalEvent: await calendarMime(a, input),
attachments: input.attachments?.map((attachment) => ({ ...attachment, content: Buffer.from("content" in attachment ? attachment.content : "", "base64") })),
})
.compile();
message.keepBcc = true;
const raw = await message.build();
await gmail(a).users.messages.send({
userId: "me",
requestBody: {
raw: raw.toString("base64url"),
threadId: input.threadId,
},
});
},
},
"gmail-imap": imapProvider(),
icloud: imapProvider(),
fastmail: imapProvider(),
yahoo: imapProvider(),
imap: imapProvider(),
};
function smtp(a: StoredAccount) {
return nodemailer.createTransport({
host: a.credentials.smtpHost,
port: a.credentials.smtpPort,
secure: a.credentials.smtpPort === 465,
requireTLS: a.credentials.smtpPort !== 465,
auth: { user: a.email, pass: a.credentials.password },
logger: false,
debug: false,
connectionTimeout: 15000,
socketTimeout: 30000,
});
}
function imapProvider() {
return {
async sent(a: StoredAccount, bound: number) {
z.number().int().min(1).max(100).parse(bound);
const client = imap(a);
try {
await imapOperation(client, () => client.connect());
const folders = await imapOperation(client, () => client.list());
const sent = folders.filter((folder) => [...folder.flags].some((flag) => flag.toLowerCase() === "\\sent"));
if (sent.length !== 1 || !sent[0].path || sent[0].path.toUpperCase() === "INBOX") throw new Error("A unique server-declared Sent folder is unavailable.");
const mailbox = sent[0].path;
const lock = await imapOperation(client, () => client.getMailboxLock(mailbox));
try {
if (!client.mailbox || client.mailbox.path !== mailbox) throw new Error("Sent mailbox identity is unavailable.");
const { exists, uidValidity, uidNext } = client.mailbox;
if (!Number.isSafeInteger(exists) || exists < 0 || !/^[1-9]\d*$/.test(String(uidValidity))) throw new Error("Sent mailbox identity is unavailable.");
const messages: Incoming[] = [];
if (exists) for await (const item of imapFetch(client, `${Math.max(1, exists - bound + 1)}:*`, { uid: true, source: imapSourceQuery, flags: true })) {
if (!item.source || !item.flags || !Number.isSafeInteger(item.uid) || item.uid < 1 || item.uid > 4294967295 || messages.length >= bound) throw new Error("Sent email source or identity is unavailable.");
const remoteId = `sent:${Buffer.from(mailbox).toString("base64url")}:${uidValidity}:${item.uid}`;
if (messages.some((m) => m.remoteId === remoteId)) throw new Error("Sent mailbox changed during retrieval.");
messages.push({ ...(await parse(a, remoteId, item.source, !item.flags.has("\\Seen"))), direction: "sent", mailbox: { path: mailbox, uidValidity: String(uidValidity), uid: item.uid } });
}
if (!client.mailbox || client.mailbox.exists !== exists || client.mailbox.uidValidity !== uidValidity || client.mailbox.uidNext !== uidNext || messages.length !== Math.min(exists, bound)) throw new Error("Sent mailbox changed during retrieval.");
return { messages, truncated: exists > bound };
} finally { lock.release(); }
} finally {
if (client.usable) await imapOperation(client, () => client.logout());
else client.close();
}
},
// INBOX is already cached completely, so search the declared archive folder
// (All Mail on Gmail IMAP) and Sent. EXAMINE, SEARCH, ENVELOPE and BODY.PEEK
// leave flags unchanged. Cancelling closes the connection.
async search(a: StoredAccount, plan: ProviderSearchPlan, { limit, signal, known }: ProviderSearchOptions): Promise<ProviderSearchFetch> {
const { groups, narrowed } = imapSearchGroups(plan);
signal.throwIfAborted();
const client = imap(a);
const cancel = () => client.close();
signal.addEventListener("abort", cancel, { once: true });
try {
await imapOperation(client, () => client.connect());
const folders = await imapOperation(client, () => client.list());
const declared = (flag: string) => {
const matches = folders.filter((folder) => [...folder.flags].some((value) => value.toLowerCase() === flag));
return matches.length === 1 && matches[0].path && matches[0].path.toUpperCase() !== "INBOX" ? matches[0].path : undefined;
};
const roles = a.provider === "gmail-imap" ? [["all", "\\all"] as const] : [["archive", "\\archive"] as const, ["sent", "\\sent"] as const];
const boxes = roles.flatMap(([role, flag]) => { const path = declared(flag); return path ? [{ role, path }] : []; });
if (!boxes.length) throw new Error("This account declares no archive or Sent folder to search.");
const messages: Incoming[] = [];
const seen = new Set<string>();
const searched: NonNullable<ProviderSearchResult["mailboxes"]> = [];
let examined = 0;
let complete = true;
for (const box of boxes) {
if (examined >= limit) { complete = false; break; }
const lock = await imapOperation(client, () => client.getMailboxLock(box.path, { readOnly: true }));
try {
if (!client.mailbox || client.mailbox.path !== box.path) throw new Error("Search mailbox identity is unavailable.");
searched.push(box.role);
const validity = String(client.mailbox.uidValidity);
// Each group holds distinct search keys; a match must satisfy every group.
let uids: Set<number> | undefined;
for (const group of groups) {
signal.throwIfAborted();
const found = await imapOperation(client, () => client.search(group, { uid: true }));
if (!found) throw new Error("The IMAP server rejected this search.");
let bounded = found;
if (found.length > IMAP_SEARCH_RESULTS_MAX) {
complete = false;
bounded = newestSearchUids(found);
}
uids = new Set(uids ? bounded.filter((uid) => uids!.has(uid)) : bounded);
}
const newest = [...uids!].filter((uid) => Number.isSafeInteger(uid) && uid > 0).sort((x, y) => y - x);
const chosen = newest.slice(0, limit - examined);
if (chosen.length < newest.length) complete = false;
examined += chosen.length;
// Sent shares the Sent import's identity so both paths keep one record.
const remoteId = (uid: number) => `${box.role === "sent" ? "sent" : "mailbox"}:${Buffer.from(box.path).toString("base64url")}:${validity}:${uid}`;
const wanted: number[] = [];
signal.throwIfAborted();
if (chosen.length) for await (const item of imapFetch(client, chosen.join(","), { uid: true, envelope: true }, { uid: true })) {
const messageId = item.envelope?.messageId;
if (!chosen.includes(item.uid) || known(`${a.id}:${remoteId(item.uid)}`, messageId) || (messageId && seen.has(messageId))) continue;
if (messageId) seen.add(messageId);
wanted.push(item.uid);
}
for (let offset = 0; offset < wanted.length; offset += 10) {
signal.throwIfAborted();
for await (const item of imapFetch(client, wanted.slice(offset, offset + 10).join(","), { uid: true, source: imapSourceQuery, flags: true }, { uid: true })) {
if (!item.source || !item.flags || !wanted.includes(item.uid)) throw new Error("Email source or identity is unavailable.");
const parsed = await syncOperation(a.provider, "parse", () => parse(a, remoteId(item.uid), item.source!, !item.flags!.has("\\Seen")));
messages.push({ ...parsed, ...(box.role === "sent" ? { direction: "sent" as const } : {}), mailbox: { path: box.path, uidValidity: validity, uid: item.uid } });
}
}
} finally { lock.release(); }
}
return { messages, examined, complete, narrowed, mailboxes: searched };
} finally {
signal.removeEventListener("abort", cancel);
if (client.usable && !signal.aborted) await imapOperation(client, () => client.logout());
else client.close();
}
},
async list(a: StoredAccount, cached: CachedMail, onBatch?: (batch: SyncBatch[]) => void, onWarning?: (warning: SyncFailure) => void, _labels?: GmailLabel[], retain = true) {
const client = imap(a);
let phase: SyncFailure["phase"] = "connect";
try {
await imapOperation(client, () => client.connect());
const lock = await imapOperation(client, () => client.getMailboxLock("INBOX"));
let changed = false;
const mailboxChanged = () => { changed = true; };
client.on("exists", mailboxChanged);
client.on("expunge", mailboxChanged);
try {
phase = "list";
const result: Incoming[] = [];
if (!client.mailbox) throw Object.assign(new Error(), { code: "INCOMPLETE_LIST" });
const { exists, uidValidity, uidNext } = client.mailbox;
const validity = String(uidValidity);
const complete = () => {
if (changed || !client.mailbox || client.mailbox.exists !== exists || String(client.mailbox.uidValidity) !== validity || client.mailbox.uidNext !== uidNext)
throw Object.assign(new Error(), { code: "INCOMPLETE_LIST" });
};
const incomplete = () => { throw Object.assign(new Error(), { code: "INCOMPLETE_LIST" }); };
if (!exists) { complete(); return result; }
const seen = new Set<number>();
const missing: number[] = [];
const oversized = new Set<number>();
let batch: SyncBatch[] = [];
// Drain flags first: ImapFlow cannot issue commands inside this iterator.
for await (const m of imapFetch(client, "1:*", { uid: true, flags: true, size: true })) {
if (!Number.isSafeInteger(m.uid) || m.uid <= 0 || seen.has(m.uid) || !m.flags) incomplete();
seen.add(m.uid);
const id = `${a.id}:${validity}:${m.uid}`;
const unread = !m.flags?.has("\\Seen");
// Summary metadata answers the flag question without reopening the
// encrypted record; a changed flag still needs the full previous row.
const summary = cached.peek?.(id);
if (summary && !retain && onBatch && summary.unread === unread) {
batch.push({ ...summary, unread });
if (batch.length === 100) { onBatch?.(batch); batch = []; }
continue;
}
const previous = cached.get(id);
if (previous) {
const incoming = { ...previous, unread };
if (retain || !onBatch) result.push(incoming);
batch.push(incoming);
if (batch.length === 100) { onBatch?.(batch); batch = []; }
} else { missing.push(m.uid); if ((m.size || 0) > rawMailLimit) oversized.add(m.uid); }
}
if (batch.length) onBatch?.(batch);
if (seen.size !== exists) incomplete();
complete();
phase = "read";
for (const uid of oversized) {
const metadata = await imapOperation(client, () => client.fetchOne(String(uid), { uid: true, envelope: true, bodyStructure: true, flags: true }, { uid: true }));
if (!metadata || metadata.uid !== uid) incomplete();
const incoming = await recoverImapMetadata(a, `${validity}:${uid}`, metadata as FetchMessageObject);
if (retain || !onBatch) result.push(incoming);
onBatch?.([incoming]); complete();
}
const ordinary = missing.filter(uid => !oversized.has(uid));
for (let offset = 0; offset < ordinary.length; offset += 2) {
const expected = new Set(ordinary.slice(offset, offset + 2));
batch = [];
for await (const m of imapFetch(client, [...expected].join(","), {
uid: true, source: imapSourceQuery, flags: true,
}, { uid: true })) {
if (!expected.delete(m.uid)) incomplete();
if (!m.source) throw Object.assign(new Error(), { code: "READ_FAILED" });
batch.push(await syncOperation(a.provider, "parse", () => parse(a, `${validity}:${m.uid}`, m.source!, !m.flags?.has("\\Seen"))));
}
if (retain || !onBatch) result.push(...batch.filter((m): m is Incoming => "body" in m));
onBatch?.(batch);
if (expected.size) incomplete();
complete();
}
await imapOperation(client, async () => {});
complete();
return result;
} finally {
client.off("exists", mailboxChanged);
client.off("expunge", mailboxChanged);
lock.release();
}
} catch (error) {
throw syncError(error, a.provider, phase);
} finally {
try {
if (client.usable) {
const loggedOut: unknown = await imapOperation(client, () => client.logout());
if (loggedOut === false) { client.close(); onWarning?.({ kind: "unknown", phase: "cleanup", code: "CLEANUP_FAILED" }); }
} else client.close();
} catch {
client.close();
onWarning?.({ kind: "unknown", phase: "cleanup", code: "CLEANUP_FAILED" });
}
}
},
async images(a: StoredAccount, m: Mail) {
return (await providers[a.provider].content(a, m)).images;
},
async content(a: StoredAccount, m: Mail) {
if (m.direction === "sent" || m.remoteId.startsWith("sent:")) throw new Error("Content recovery for this Sent email is unsupported. Sync Sent again.");
// An archived UID belongs to the old INBOX, never to the archive folder.
if (m.archived) throw new Error("Content is unavailable for this older archived IMAP email.");
const identifiers = /^(\d+):([1-9]\d*)$/.exec(m.remoteId);
if (!identifiers) throw new Error("Email identifiers are unavailable. Sync and try again.");
const client = imap(a);
try {
await client.connect();
const lock = await client.getMailboxLock("INBOX");
try {
if (!client.mailbox || String(client.mailbox.uidValidity) !== identifiers[1])
throw new Error("Mailbox identifiers changed. Sync before loading content.");
const result = await client.fetchOne(identifiers[2], { source: imapSourceQuery }, { uid: true });
if (!result || !result.source) throw new Error("Email source is unavailable.");
const parsed = await parse(a, m.remoteId, result.source, m.unread);
if (m.messageId && parsed.messageId !== m.messageId)
throw new Error("Email identity changed. Content was not loaded.");
return { html: parsed.html ?? null, images: parsed.images!, attachments: parsed.attachments, ...(parsed.invitation ? { invitation: parsed.invitation } : {}) };
} finally {
lock.release();
}
} finally {
if (client.usable) await client.logout();
else client.close();
}
},
async update(a: StoredAccount, m: Mail, action: "archive" | "read" | "unread" | "inbox" | "trash") {
if (m.direction === "sent" || m.remoteId.startsWith("sent:") || (m.mailbox && !m.archived)) throw new Error("Changing this Sent email is unsupported. Change it in your provider.");
const identifiers = /^(\d+):([1-9]\d*)$/.exec(m.remoteId);
if (action === "trash" && (!identifiers ||
identifiers.slice(1).some((value) => !Number.isSafeInteger(Number(value)) || Number(value) < 1 || Number(value) > 4294967295)))
throw new Error("Email identifiers are unavailable. Sync before moving this email to Trash.");
const client = imap(a);
let movedToTrash = false;
let restoredIdentity: { remoteId?: string; mailbox?: Mail["mailbox"] } | undefined;
try {
await client.connect();
// ImapFlow otherwise falls back to COPY, Deleted and EXPUNGE.
const rev2 = client.enabled.has("IMAP4REV2") ||
(client.capabilities.has("IMAP4rev2") && !client.capabilities.has("IMAP4rev1"));
let trash: string | undefined;
let trashSource = "INBOX";
if (action === "trash") {
if (!client.capabilities.has("MOVE") && !rev2)
throw new Error("This server does not support safe Move to Trash. Move this email in your provider.");
const folders = await client.list();
// specialUse can be guessed from names and hides duplicate server flags.
const withFlag = (flag: string) => folders.filter((folder) =>
[...folder.flags].some((value) => value.toLowerCase() === flag));
const targets = withFlag("\\trash");
if (targets.length !== 1 || !targets[0].path)
throw new Error("A unique Trash folder is unavailable. Move this email in your provider.");
trash = targets[0].path;
if (m.archived) {
const sources = withFlag(a.provider === "gmail-imap" ? "\\all" : "\\archive");
if (sources.length !== 1 || !sources[0].path)
throw new Error("A unique archive folder is unavailable. Move this email in your provider.");
trashSource = sources[0].path;
}
if (trash === trashSource || trash.toUpperCase() === "INBOX")
throw new Error("The Trash folder is unsafe. Move this email in your provider.");
}
let restoreSource: string | undefined;
if (action === "inbox") {
if (!m.archived) throw new Error("This email is already in Inbox.");
// Gmail restores by copying into INBOX (adding the label); removing mail from All Mail can delete it.
if ((a.provider !== "gmail-imap" && !client.capabilities.has("MOVE")) || !client.capabilities.has("UIDPLUS"))
throw new Error("This server does not support safe Move to Inbox with confirmed identifiers. Move it in your provider.");
const sources = (await client.list()).filter(folder => [...folder.flags].some(flag => flag.toLowerCase() === (a.provider === "gmail-imap" ? "\\all" : "\\archive")));
if (sources.length !== 1 || !sources[0].path || sources[0].path.toUpperCase() === "INBOX")
throw new Error("A unique archive folder is unavailable. Move this email in your provider.");
restoreSource = sources[0].path;
}
// A confirmed archive move records the exact destination; open it first.
const mailbox = action === "inbox" ? (m.mailbox?.path || restoreSource) : action === "trash" ? (m.mailbox?.path || trashSource) : m.archived && (action === "read" || action === "unread")
? (m.mailbox?.path || (await client.list()).find((folder) => folder.specialUse === (a.provider === "gmail-imap" ? "\\All" : "\\Archive"))?.path || (a.provider === "gmail-imap" ? null : "Archive"))
: "INBOX";
if (!mailbox) throw new Error("Gmail All Mail is unavailable. Enable it in IMAP before marking archived mail read.");
const lock = await client.getMailboxLock(mailbox);
try {
if ((action === "trash" || action === "inbox") && (!client.mailbox ||
(mailbox === "INBOX" ? client.mailbox.path.toUpperCase() !== "INBOX" : client.mailbox.path !== mailbox)))
throw new Error("Email mailbox changed. It was not moved to Trash.");
const [validity, originalUid] = m.remoteId.split(":");
let uid = originalUid;
if (m.archived && (action === "read" || action === "unread" || action === "inbox" || action === "trash")) {
if (!m.messageId) throw new Error("This archived email has no Message-ID. Change it in your provider.");
if ((action === "trash" || action === "inbox") && (m.messageId.length > 998 || /[\s\u0000-\u001f\u007f]/.test(m.messageId)))
throw new Error("This archived email has an invalid Message-ID. Move it in your provider.");
// The binding from the confirmed archive move names the destination
// exactly; it is valid only inside the same mailbox and validity.
const bound = Boolean(m.mailbox && client.mailbox && client.mailbox.path === m.mailbox.path && String(client.mailbox.uidValidity) === m.mailbox.uidValidity);
if (bound) uid = String(m.mailbox!.uid);
else {
// Moving IMAP mail changes its UID; never reuse the old inbox UID in the archive.
const matches = await client.search({ header: { "message-id": m.messageId } }, { uid: true });
if (!matches || matches.length !== 1)
throw new Error("Could not identify this archived email. Change it in your provider.");
uid = String(matches[0]);
}
if ((action === "trash" || action === "inbox") && (!/^[1-9]\d*$/.test(uid) || !Number.isSafeInteger(Number(uid)) || Number(uid) > 4294967295))
throw new Error("Could not identify this archived email. Move it in your provider.");
const candidate = await client.fetchOne(uid, (action === "trash" || action === "inbox") ? { envelope: true, headers: ["message-id"] } : { headers: ["message-id"] }, { uid: true });
if (!candidate || ((action === "trash" || action === "inbox") && String(candidate.uid) !== uid) || !candidate.headers || (await simpleParser(candidate.headers)).messageId !== m.messageId)
throw new Error("Could not identify this archived email. Change it in your provider.");
// A recovered match without a move binding is not identity for a
// destructive move unless the Subject and any available Date agree.
if (!bound && (action === "trash" || action === "inbox") && ((candidate.envelope?.date && candidate.envelope.date.getTime() !== m.date) || (candidate.envelope?.subject || "(No subject)") !== m.subject))
throw new Error("Could not confirm this archived email. Change it in your provider.");
} else if (
!client.mailbox ||
String(client.mailbox.uidValidity) !== validity
)
throw new Error(
"Mailbox identifiers changed. Sync before changing this email.",
);
if (action === "trash") {
if (!m.archived && m.messageId) {
const candidate = await client.fetchOne(uid, { headers: ["message-id"] }, { uid: true });
if (!candidate || String(candidate.uid) !== uid || !candidate.headers || (await simpleParser(candidate.headers)).messageId !== m.messageId)
throw new Error("Email identity changed. It was not moved to Trash.");
}
if (!(await client.messageMove(uid, trash!, { uid: true })))
throw new Error("Move to Trash was not confirmed. Check your provider before trying again.");
movedToTrash = true;
} else if (action === "inbox") {
const moved = a.provider === "gmail-imap"
? await client.messageCopy(uid, "INBOX", { uid: true })
: await client.messageMove(uid, "INBOX", { uid: true });
if (!moved) throw new Error("Move to Inbox was not confirmed. Check your provider before trying again.");
movedToTrash = true; // Confirmed moves must not become retryable on cleanup failure.
const targetUid = moved.uidMap?.get(Number(uid));
if (!targetUid || !Number.isSafeInteger(targetUid) || targetUid < 1 || targetUid > 4294967295 || !/^[1-9]\d*$/.test(String(moved.uidValidity)))
throw new Error("Email moved to Inbox, but its new identifiers are unavailable. Sync before changing it again.");
restoredIdentity = { remoteId: `${moved.uidValidity}:${targetUid}` };
} else if (action === "unread") {
if (!(await client.messageFlagsRemove(uid, ["\\Seen"], { uid: true })))
throw new Error("Could not mark this email as unread. Sync and try again.");
} else if (action === "read") {
if (!(await client.messageFlagsAdd(uid, ["\\Seen"], { uid: true })))
throw new Error(
"Could not mark this email as read. Sync and try again.",
);
} else {
// A COPY + Deleted + EXPUNGE fallback can expunge unrelated mail.
if (!client.capabilities.has("MOVE") && !rev2)
throw new Error("This server does not support safe Archive. Archive this email in your provider.");
const folders = await client.list();
const googleImap = a.provider === "gmail-imap";
const archive =
folders.find((f) => f.specialUse === (googleImap ? "\\All" : "\\Archive"))?.path ||
(googleImap ? null : "Archive");
if (!archive)
throw new Error("Gmail All Mail is unavailable. Enable it in IMAP before archiving.");
if (!folders.some((f) => f.path === archive))
await client.mailboxCreate(archive);
const moved = await client.messageMove(uid, archive, { uid: true });
if (!moved)
throw new Error(
"Could not archive this email. Sync and try again.",
);
// Record the confirmed destination binding so archived actions can
// target this exact message instead of searching by Message-ID.
const targetUid = moved.uidMap?.get(Number(uid));
if (targetUid && Number.isSafeInteger(targetUid) && targetUid >= 1 && targetUid <= 4294967295 && /^[1-9]\d*$/.test(String(moved.uidValidity)))
restoredIdentity = { mailbox: { path: archive, uidValidity: String(moved.uidValidity), uid: targetUid } };
}
} finally {
try { lock.release(); }
catch (error) { if (!movedToTrash) throw error; }
}
} finally {
try {
if (client.usable) await client.logout();
else client.close();
} catch (error) {
if (!movedToTrash) throw error;
// A confirmed move must not become retryable because cleanup failed.
try { client.close(); } catch {}
}
}
return restoredIdentity;
},
async send(a: StoredAccount, input: SendInput) {
input = sendSchema.parse(input);
if (input.attachments?.some(file => "id" in file)) return sendReferenced(a, input);
const raw = await new MailComposer({
from: a.email,
to: input.to,
cc: input.cc,
subject: input.subject,
text: input.body,
html: input.html,
inReplyTo: input.inReplyTo,
references: input.references || input.inReplyTo,
icalEvent: await calendarMime(a, input),
attachments: input.attachments?.map((attachment) => ({ ...attachment, content: Buffer.from("content" in attachment ? attachment.content : "", "base64") })),
})
.compile()
.build();
const receipt = await smtp(a).sendMail({
envelope: { from: a.email, to: [...input.to, ...(input.cc || []), ...(input.bcc || [])], size: raw.length },
raw,
});
const warnings: string[] = [];
if (receipt.rejected?.length) {
if (receipt.accepted?.length === 0)
throw Object.assign(new Error("SMTP rejected every recipient."), { code: "EENVELOPE" });
warnings.push("some recipients were rejected");
}
// Gmail SMTP saves Sent itself; appending over IMAP would duplicate it.
if (a.provider === "gmail-imap") {
if (warnings.length) throw new SentCopyError(`Email was sent, but ${warnings.join(" and ")}. Do not resend.`);
return;
}
await saveSentCopy(a, raw, warnings);
},
};
}
export const attachmentSchema = z.object({
filename: z.string().min(1).max(255).refine((value) => !/[\u0000-\u001f\u007f/\\:]/.test(value) && value !== "." && value !== "..", "Attachment filename is unsafe"),
contentType: z.string().max(127).regex(/^[\w!#$&^.+-]+\/[\w!#$&^.+-]+$/),
content: z.string().max(699052).refine((value) => { const size = canonicalBase64Size(value); return size !== null && size <= 524288; }, "Attachment content must be canonical base64 of at most 512 KiB"),
}).strict();
export const attachmentMetadataSchema = attachmentSchema.omit({ content: true });
export const attachmentReferenceSchema = attachmentMetadataSchema.extend({ id: z.uuid(), size: z.number().int().min(0).max(ATTACHMENT_BYTES) }).strict();
export const attachmentsSchema = z.array(z.union([attachmentSchema, attachmentReferenceSchema])).max(ATTACHMENT_COUNT).refine((files) => files.reduce((n, file) => n + attachmentSize(file), 0) <= ATTACHMENT_BYTES, "Attachments exceed 25 MiB total").refine(files => files.reduce((n, file) => n + ("content" in file ? attachmentSize(file) : 0), 0) <= 1048576, "Inline legacy attachments exceed 1 MiB total");
export const sendSchema = z.object({
accountId: z.string(),
calendarDraft: z.object({ id: z.uuid(), revision: z.number().int().min(1) }).strict().optional(),
to: z.array(z.email().max(320)).min(1).max(100),
bcc: z.array(z.email().max(320)).max(100).optional(),
cc: z.array(z.email().max(320)).max(100).optional(),
attachments: attachmentsSchema.optional(),
references: z
.array(
z
.string()
.max(998)
.refine((v) => !/[\u0000-\u001f\u007f]/.test(v)),
)
.max(100)
.optional(),
subject: z
.string()
.max(998)
.refine((v) => !/[\u0000-\u001f\u007f]/.test(v), "Subject must be one line"),
body: z.string().min(1).max(1000000),
html: z.string().max(COMPOSE_HTML_LIMIT).optional(),
inReplyTo: z
.string()
.max(998)
.refine((v) => !/[\u0000-\u001f\u007f]/.test(v))
.optional(),
threadId: z.string().max(100).optional(),
}).refine(validComposeMarkup, "Formatted content must match its plain text alternative").refine((input) => Buffer.byteLength(JSON.stringify(input), "utf8") <= 1900000, "Message exceeds the request size limit");
const { html: _richHtml, ...plainSendShape } = sendSchema.shape;
export const plainSendSchema = z.object({ ...plainSendShape, to: z.union([plainSendShape.to, z.email().max(320).transform(value => [value])]) }).strict().refine((input) => Buffer.byteLength(JSON.stringify(input), "utf8") <= 1900000, "Message exceeds the request size limit");
export type SendInput = z.infer<typeof sendSchema>;
// Resolve encrypted, reviewed metadata instead of accepting calendar text from a send caller.
export async function calendarMime(account: StoredAccount, input: SendInput) {
if (!input.calendarDraft) return undefined;
const { get } = await import("./store.js");
const draft = get("draft", input.calendarDraft.id);
if (!draft || draft.accountId !== account.id || draft.revision !== input.calendarDraft.revision || !draft.sendUnknown || !draft.invitation ||
!isDeepStrictEqual(Array.isArray(draft.to) ? draft.to : [draft.to], input.to) || !isDeepStrictEqual(draft.cc || [], input.cc || []) || !isDeepStrictEqual(draft.bcc || [], input.bcc || []) || draft.subject !== input.subject || draft.body !== input.body || draft.html !== input.html)
throw new Error("Invitation draft changed. Recover the saved draft before sending.");
const context = draft.calendarContext;
const event = context && get("calendarEvent", context.eventId);
if (!context || !event || event.accountId !== account.id || event.revision !== context.revision || event.sequence !== draft.invitation.sequence) throw new Error("Calendar event changed. Review it before sending this draft.");
if (context.sourceMailId) {
const mail = get("mail", context.sourceMailId);
const fingerprint = mail && createHash("sha256").update(JSON.stringify([mail.id, mail.accountId, mail.remoteId, mail.messageId, mail.address, mail.invitation])).digest("hex");
if (!mail || mail.accountId !== account.id || fingerprint !== context.sourceFingerprint) throw new Error("Invitation source changed. Reopen it before replying.");
}
if (!isDeepStrictEqual(draft.attachments || [], input.attachments || [])) throw new Error("Invitation attachments changed. Recover the draft before sending.");
const invitation = draft.invitation;
const own = account.email.toLowerCase();
if (invitation.method === "REPLY" ? invitation.attendees.length !== 1 || invitation.attendees[0].email.toLowerCase() !== own : invitation.organizer.toLowerCase() !== own)
throw new Error("The invitation belongs to another account.");
return { method: invitation.method, filename: "invitation.ics", content: generateICalendar(invitation, { stamp: new Date().toISOString().replace(/\.\d{3}Z$/, ".000Z"), replyingAttendee: invitation.method === "REPLY" ? account.email : undefined }) };
}
// Saves an accepted SMTP message to Sent. Failure after delivery is a warning, never a resend.
async function saveSentCopy(a: StoredAccount, raw: Buffer, warnings: string[] = []) {
let client: ImapFlow | undefined;
try {
client = imap(a);
await client.connect();
const folders = await client.list();
const sent =
folders.find((f) => f.specialUse === "\\Sent")?.path || "Sent";
if (!folders.some((f) => f.path === sent))
await client.mailboxCreate(sent);
if (!(await client.append(sent, raw, ["\\Seen"])))
throw new Error("Sent copy was not saved.");
} catch {
warnings.push("saving a Sent copy failed");
} finally {
try {
if (client?.usable) {
if (Object.is(await client.logout(), false))
warnings.push("closing the Sent-copy connection failed");
} else client?.close();
} catch {
warnings.push("closing the Sent-copy connection failed");
}
}
if (warnings.length)
throw new SentCopyError(
`Email was sent, but ${warnings.join(" and ")}. Do not resend.`,
);
}
// Referenced envelopes are spooled as encrypted chunks, measured, then replayed.
// The SMTP transport passes envelope.size to its connection, which rejects an
// advertised SIZE overflow before MAIL FROM on that same connection.
async function sendReferenced(a: StoredAccount, input: SendInput) {
let admitted = false;
return attachmentWork(async () => {
admitted = true;
const { uploadBytes, eraseUpload, account } = await import("./store.js");
const message = new MailComposer({
from: a.email, to: input.to, cc: input.cc, ...(a.provider === "gmail" ? { bcc: input.bcc } : {}),
subject: input.subject, text: input.body, html: input.html,
inReplyTo: input.inReplyTo, references: input.references || input.inReplyTo,
icalEvent: await calendarMime(a, input), attachments: await mimeAttachments(a.id, input.attachments),
}).compile();
message.keepBcc = a.provider === "gmail";
const signal = AbortSignal.timeout(60000);
const source = message.createReadStream();
const abort = () => source.destroy(new Error("Attachment preparation timed out."));
signal.addEventListener("abort", abort, { once: true });
let spool: { id: string; size: number } | undefined;
let submitted = false, accepted = false, partialRejection = false;
try {
spool = await storeAttachmentBytes(a.id, "message.eml", "message/rfc822", source, signal, MIME_BYTES);
const current = account(a.id);
if (current.provider !== a.provider || current.email !== a.email || !isDeepStrictEqual(current.credentials, a.credentials)) throw new AttachmentPreflightError("Sending account changed before submission.");
if (a.provider === "gmail") {
const auth = oauth();
auth.setCredentials({ refresh_token: a.credentials.refreshToken });
auth.transporter.defaults.retry = false;
const data = Readable.from(gmailJson(uploadBytes(spool.id), input.threadId));
submitted = true;
try {
await auth.request({ url: "https://gmail.googleapis.com/gmail/v1/users/me/messages/send", method: "POST",
headers: { "Content-Type": "application/json" }, data, timeout: 60000, retry: false });
accepted = true;
} finally { data.destroy(); }
} else {
const transport = smtp(a);
const raw = Readable.from(uploadBytes(spool.id));
try {
submitted = true;
const receipt = await transport.sendMail({ envelope: { from: a.email, to: [...input.to, ...(input.cc || []), ...(input.bcc || [])], size: spool.size }, raw });
if (receipt?.rejected?.length && receipt.accepted?.length === 0)
throw Object.assign(new Error("SMTP rejected every recipient."), { code: "EENVELOPE" });
accepted = true;
partialRejection = Boolean(receipt?.rejected?.length);
} finally { raw.destroy(); transport.close(); }
// ImapFlow APPEND needs one Buffer. The spool is already bounded by
// MIME_BYTES, so read it once for the Sent copy before it is erased.
if (a.provider !== "gmail-imap") {
let copy: Buffer | undefined;
try { const chunks: Buffer[] = []; for await (const chunk of uploadBytes(spool.id)) chunks.push(chunk); copy = Buffer.concat(chunks); }
catch { throw new SentCopyError("Email was sent, but saving a Sent copy failed. Do not resend."); }
await saveSentCopy(a, copy, partialRejection ? ["some recipients were rejected"] : []);
} else if (partialRejection) throw new SentCopyError("Email was sent, but some recipients were rejected. Do not resend.");
}
} catch (error) {
if (accepted && !(error instanceof SentCopyError)) throw new SentCopyError("Email was sent, but closing its attachment stream failed. Do not resend.");
const failure = error as { code?: string; command?: string; message?: string };
if (!submitted || (failure.code === "EMESSAGE" && failure.command === "MAIL FROM" && /size larger than allowed/i.test(failure.message || "")))
throw new AttachmentPreflightError(error instanceof Error ? error.message : "Attachment preparation failed before submission.");
throw error;
} finally {
signal.removeEventListener("abort", abort); source.destroy();
if (spool) {
try { eraseUpload(spool.id); }
catch (error) { if (accepted) throw new SentCopyError("Email was sent, but temporary attachment cleanup failed. Do not resend."); throw error; }
}
}
}).catch(error => {
if (!admitted) throw new AttachmentPreflightError("Attachment work is busy. Nothing was submitted. Try again after the current operation.");
throw error;
});
}
const metadataBody = "Message body was not parsed because the source exceeds the automatic import limit. Attachments can be downloaded separately.";
function safeHeader(value: string) { return value.replace(/[\r\n\0]/g, " ").slice(0, 8192); }
async function metadataIncoming(a: StoredAccount, remoteId: string, headers: { name?: string | null; value?: string | null }[], files: MailAttachment[], unread: boolean, date?: number) {
const kept = new Set(["from", "to", "cc", "reply-to", "subject", "date", "message-id", "in-reply-to", "references"]);
const source = headers.filter(h => h.name && kept.has(h.name.toLowerCase())).slice(0, 64)
.map(h => `${h.name}: ${safeHeader(h.value || "")}`).join("\r\n");
const parsed = await parse(a, remoteId, Buffer.from(`${source}\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n${metadataBody}`), unread, date);
return { ...parsed, bodyUnavailable: true, attachments: files, html: null, images: [] };
}
// Exclude body.data from Gmail's full-message response. The response retains only
// attachment locators and bounded headers, even for a huge message.
export const gmailMetadataFields = (() => {
let part = "partId,mimeType,filename,headers,body(attachmentId,size)";
for (let depth = 0; depth < 16; depth++) part = `partId,mimeType,filename,headers,body(attachmentId,size),parts(${part})`;
return `id,threadId,labelIds,internalDate,payload(${part})`;
})();
export async function recoverGmailMetadata(a: StoredAccount, remoteId: string, api = gmail(a)) {
const result = await api.users.messages.get({ userId: "me", id: remoteId, format: "full", fields: gmailMetadataFields },
{ timeout: 30000, retry: false, maxContentLength: 2 * 1024 * 1024 });
if (result.data.id !== remoteId) throw new Error("Email identity changed.");
const files: MailAttachment[] = [];
const walk = (part: gmail_v1.Schema$MessagePart, depth: number) => {
if (depth > 16 || files.length >= 128) throw new Error("Attachment metadata limit exceeded.");
if (part.filename && part.body?.attachmentId && typeof part.body.size === "number") files.push({
id: createHash("sha256").update(`${a.id}:${remoteId}:gmail:${part.body.attachmentId}`).digest("hex"),
filename: part.filename.replace(/[\u0000-\u001f\u007f/\\]/g, "_").slice(0, 255), contentType: part.mimeType || "application/octet-stream", size: part.body.size, remotePart: part.body.attachmentId, unavailable: true,
});
for (const child of part.parts || []) walk(child, depth + 1);
};
if (result.data.payload) walk(result.data.payload, 0);
return { ...(await metadataIncoming(a, remoteId, result.data.payload?.headers || [], files, result.data.labelIds?.includes("UNREAD") || false, Number(result.data.internalDate))), threadId: result.data.threadId || undefined, labelIds: result.data.labelIds || [] };
}
export async function recoverImapMetadata(a: StoredAccount, remoteId: string, message: FetchMessageObject) {
const files: MailAttachment[] = [];
let nodes = 0;
const walk = (part: MessageStructureObject, depth: number) => {
if (++nodes > 128 || depth > 16) throw new Error("Attachment metadata limit exceeded.");
const name = part.dispositionParameters?.filename || part.parameters?.name;
if (part.part && name) {
const wireSize = part.size || 0;
files.push({ id: createHash("sha256").update(`${a.id}:${remoteId}:imap:${part.part}`).digest("hex"),
filename: name.replace(/[\u0000-\u001f\u007f/\\]/g, "_").slice(0, 255), contentType: part.type || "application/octet-stream",
size: part.encoding?.toLowerCase() === "base64" ? Math.ceil(wireSize * 3 / 4) : wireSize,
sizeIsEstimate: true, remotePart: part.part, unavailable: true });
}
for (const child of part.childNodes || []) walk(child, depth + 1);
};
if (message.bodyStructure) walk(message.bodyStructure, 0);
const env = message.envelope;
const addresses = (list: NonNullable<FetchMessageObject["envelope"]>["from"]) => (list || []).map(x => `${x.name || ""} <${x.address || ""}>`).join(", ");
return metadataIncoming(a, remoteId, [{ name: "From", value: addresses(env?.from) }, { name: "To", value: addresses(env?.to) }, { name: "Cc", value: addresses(env?.cc) },
{ name: "Subject", value: env?.subject }, { name: "Message-ID", value: env?.messageId }, { name: "Date", value: env?.date?.toUTCString() }], files, !message.flags?.has("\\Seen"));
}