();
+ for (const message of messages) {
+ const ics = icsFor.get(message);
+ if (ics !== undefined) resolved.set(message.id, ics);
+ }
+ return extractCalendarReplies(messages, resolved);
}
const declined = [
@@ -944,7 +1136,7 @@ describe("extractCalendarReplies", () => {
].join("\r\n");
it("extracts a decline", () => {
- const [reply] = extractCalendarReplies([replyMessage(declined)]);
+ const [reply] = extractReplies([replyMessage(declined)]);
expect(reply).toMatchObject({
messageId: "m1",
uid: "uid-1@google.com",
@@ -961,20 +1153,20 @@ describe("extractCalendarReplies", () => {
it("extracts accepted and tentative", () => {
const accepted = declined.replace("PARTSTAT=DECLINED", "PARTSTAT=ACCEPTED");
const tentative = declined.replace("PARTSTAT=DECLINED", "PARTSTAT=TENTATIVE");
- expect(extractCalendarReplies([replyMessage(accepted)])[0].partstat).toBe("ACCEPTED");
- expect(extractCalendarReplies([replyMessage(tentative)])[0].partstat).toBe("TENTATIVE");
+ expect(extractReplies([replyMessage(accepted)])[0].partstat).toBe("ACCEPTED");
+ expect(extractReplies([replyMessage(tentative)])[0].partstat).toBe("TENTATIVE");
});
it("ignores NEEDS-ACTION", () => {
const pending = declined.replace("PARTSTAT=DECLINED", "PARTSTAT=NEEDS-ACTION");
- expect(extractCalendarReplies([replyMessage(pending)])).toEqual([]);
+ expect(extractReplies([replyMessage(pending)])).toEqual([]);
});
it("ignores non-REPLY methods and messages with no ICS", () => {
const request = declined.replace("METHOD:REPLY", "METHOD:REQUEST");
- expect(extractCalendarReplies([replyMessage(request)])).toEqual([]);
+ expect(extractReplies([replyMessage(request)])).toEqual([]);
expect(
- extractCalendarReplies([
+ extractReplies([
{ ...replyMessage(declined), payload: part("text/plain", { data: "hi" }) },
])
).toEqual([]);
@@ -985,7 +1177,7 @@ describe("extractCalendarReplies", () => {
"END:VEVENT",
"RECURRENCE-ID:20260804T140000Z\r\nEND:VEVENT"
);
- const [reply] = extractCalendarReplies([replyMessage(ics)]);
+ const [reply] = extractReplies([replyMessage(ics)]);
expect(reply.occurrence?.toISOString()).toBe("2026-08-04T14:00:00.000Z");
expect(reply.allDay).toBe(false);
});
@@ -995,7 +1187,7 @@ describe("extractCalendarReplies", () => {
"END:VEVENT",
"RECURRENCE-ID;VALUE=DATE:20260804\r\nEND:VEVENT"
);
- const [reply] = extractCalendarReplies([replyMessage(ics)]);
+ const [reply] = extractReplies([replyMessage(ics)]);
expect(reply.occurrence?.toISOString()).toBe("2026-08-04T00:00:00.000Z");
expect(reply.allDay).toBe(true);
});
@@ -1007,7 +1199,7 @@ describe("extractCalendarReplies", () => {
"END:VEVENT",
"RECURRENCE-ID;TZID=America/Toronto:20260804T100000\r\nEND:VEVENT"
);
- const [reply] = extractCalendarReplies([replyMessage(ics)]);
+ const [reply] = extractReplies([replyMessage(ics)]);
expect(reply.occurrence?.toISOString()).toBe("2026-08-04T10:00:00.000Z");
});
@@ -1016,7 +1208,7 @@ describe("extractCalendarReplies", () => {
"END:VEVENT",
"COMMENT:Could we move this to Thursday?\r\nEND:VEVENT"
);
- expect(extractCalendarReplies([replyMessage(ics)])[0].comment).toBe(
+ expect(extractReplies([replyMessage(ics)])[0].comment).toBe(
"Could we move this to Thursday?"
);
});
@@ -1026,7 +1218,7 @@ describe("extractCalendarReplies", () => {
"END:VEVENT",
"COMMENT:Could we move this\r\n to Thursday?\r\nEND:VEVENT"
);
- expect(extractCalendarReplies([replyMessage(ics)])[0].comment).toBe(
+ expect(extractReplies([replyMessage(ics)])[0].comment).toBe(
"Could we move this to Thursday?"
);
});
@@ -1036,7 +1228,7 @@ describe("extractCalendarReplies", () => {
"END:VEVENT",
"COMMENT:Line one\\nLine two\\, and more\r\nEND:VEVENT"
);
- expect(extractCalendarReplies([replyMessage(ics)])[0].comment).toBe(
+ expect(extractReplies([replyMessage(ics)])[0].comment).toBe(
"Line one\nLine two, and more"
);
});
@@ -1046,7 +1238,7 @@ describe("extractCalendarReplies", () => {
"END:VEVENT",
"COMMENT:Back\\\\nslash\r\nEND:VEVENT"
);
- expect(extractCalendarReplies([replyMessage(ics)])[0].comment).toBe(
+ expect(extractReplies([replyMessage(ics)])[0].comment).toBe(
"Back\\nslash"
);
});
@@ -1056,7 +1248,7 @@ describe("extractCalendarReplies", () => {
"CN=Beth Round:",
'CN=Beth Round;X-RESPONSE-COMMENT="Sorry, conflict":'
);
- expect(extractCalendarReplies([replyMessage(ics)])[0].comment).toBe(
+ expect(extractReplies([replyMessage(ics)])[0].comment).toBe(
"Sorry, conflict"
);
});
@@ -1066,7 +1258,7 @@ describe("extractCalendarReplies", () => {
"CN=Beth Round:",
'CN=Beth Round;X-RESPONSE-COMMENT="Back by 3:00":'
);
- const [reply] = extractCalendarReplies([replyMessage(ics)]);
+ const [reply] = extractReplies([replyMessage(ics)]);
expect(reply.comment).toBe("Back by 3:00");
expect(reply.attendeeEmail).toBe("beth@example.test");
});
@@ -1076,28 +1268,28 @@ describe("extractCalendarReplies", () => {
"Beth Round has declined this invitation with a note:
" +
'"Could we move this to Thursday?"
' +
'';
- expect(extractCalendarReplies([replyMessage(declined, { html })])[0].comment).toBe(
+ expect(extractReplies([replyMessage(declined, { html })])[0].comment).toBe(
"Could we move this to Thursday?"
);
});
it("returns null comment when no source has one", () => {
const html = "Beth Round has declined this invitation.
";
- expect(extractCalendarReplies([replyMessage(declined, { html })])[0].comment).toBeNull();
+ expect(extractReplies([replyMessage(declined, { html })])[0].comment).toBeNull();
});
it("falls back to the From display name when CN is absent", () => {
const ics = declined.replace(";CN=Beth Round:", ":");
const msg = replyMessage(ics);
msg.payload.headers = [{ name: "From", value: '"Beth Round" ' }];
- expect(extractCalendarReplies([msg])[0].attendeeName).toBe("Beth Round");
+ expect(extractReplies([msg])[0].attendeeName).toBe("Beth Round");
});
it("returns one descriptor per reply message in the conversation", () => {
const second = declined
.replace("PARTSTAT=DECLINED", "PARTSTAT=ACCEPTED")
.replace("beth@example.test", "sam@example.test");
- const replies = extractCalendarReplies([
+ const replies = extractReplies([
replyMessage(declined, { id: "m1" }),
replyMessage(second, { id: "m2" }),
]);
@@ -1110,8 +1302,8 @@ describe("extractCalendarReplies", () => {
it("skips a reply with no resolvable UID or attendee email", () => {
const noUid = declined.replace("UID:uid-1@google.com\r\n", "");
const noAttendee = declined.replace(/^ATTENDEE.*\r\n/m, "");
- expect(extractCalendarReplies([replyMessage(noUid)])).toEqual([]);
- expect(extractCalendarReplies([replyMessage(noAttendee)])).toEqual([]);
+ expect(extractReplies([replyMessage(noUid)])).toEqual([]);
+ expect(extractReplies([replyMessage(noAttendee)])).toEqual([]);
});
});
diff --git a/connectors/google/src/mail/gmail-api.ts b/connectors/google/src/mail/gmail-api.ts
index eabe880f..71e8464e 100644
--- a/connectors/google/src/mail/gmail-api.ts
+++ b/connectors/google/src/mail/gmail-api.ts
@@ -791,14 +791,100 @@ function icsProp(ics: string, name: string): string | null {
return m ? m[2].trim() : null;
}
+/**
+ * MIME types an iCalendar part is delivered under. Google and Exchange use
+ * `text/calendar`; a sizeable minority of senders use `application/ics`.
+ */
+const CALENDAR_MIME_TYPES = new Set(["text/calendar", "application/ics"]);
+
+/**
+ * True when a payload part carries an iCalendar body. A few senders label the
+ * part `application/octet-stream` (or even `text/plain`) and only the `.ics`
+ * filename gives it away, so the filename is consulted as well.
+ */
+function isCalendarPart(part: GmailMessagePart): boolean {
+ const mimeType = (part.mimeType ?? "").split(";")[0].trim().toLowerCase();
+ if (CALENDAR_MIME_TYPES.has(mimeType)) return true;
+ return /\.ics$/i.test(part.filename ?? "");
+}
+
+/**
+ * True when a message carries a calendar part. Lets a caller skip the cost of
+ * resolving an API client for a batch of ordinary mail.
+ */
+export function hasCalendarPart(message: GmailMessage): boolean {
+ return findCalendarPart(message.payload) !== null;
+}
+
+/** First calendar part anywhere in the payload tree, or null. */
+function findCalendarPart(part: GmailMessagePart): GmailMessagePart | null {
+ if (isCalendarPart(part)) return part;
+ for (const child of part.parts ?? []) {
+ const found = findCalendarPart(child);
+ if (found) return found;
+ }
+ return null;
+}
+
+/**
+ * Reads each message's iCalendar body, keyed by message id.
+ *
+ * Gmail treats a calendar part as an attachment even when the sender inlined
+ * it: it synthesizes a filename (`invite.ics`) and moves the body out to
+ * `attachmentId`, leaving `body.data` empty. The ICS behind an invitation,
+ * update, cancellation or RSVP is therefore absent from the payload
+ * `messages.get` returns, and reading it costs a second
+ * `messages.attachments.get` call — which is why this step is separate from
+ * the pure parsing that consumes it.
+ *
+ * Ordinary mail issues no request: a message with no calendar part is skipped
+ * before any fetch. A fetch that fails leaves the message out of the map
+ * rather than aborting the pass, so the conversation degrades to syncing as a
+ * plain email.
+ */
+export async function resolveIcsByMessage(
+ api: Pick,
+ messages: GmailMessage[]
+): Promise> {
+ const resolved = new Map();
+
+ for (const message of messages) {
+ const calendarPart = findCalendarPart(message.payload);
+ if (!calendarPart) continue;
+
+ // Inline body: rare in practice, but free when it happens.
+ if (calendarPart.body?.data) {
+ resolved.set(message.id, decodeBase64Url(calendarPart.body.data));
+ continue;
+ }
+
+ const attachmentId = calendarPart.body?.attachmentId;
+ if (!attachmentId) continue;
+
+ try {
+ const { data } = await api.getAttachment(message.id, attachmentId);
+ if (data) resolved.set(message.id, decodeBase64Url(data));
+ } catch (error) {
+ console.warn(
+ `[gmail] could not read the calendar part of message ${message.id}:`,
+ error
+ );
+ }
+ }
+
+ return resolved;
+}
+
/**
* Classify a Gmail conversation's relationship to a calendar event for bundling.
* Two signals: our own `X-Plot-Event-UID` header (a Plot-sent reply chain), or a
- * `text/calendar` part (invitation/update/cancellation/RSVP). Only updates,
+ * calendar part (invitation/update/cancellation/RSVP), whose body arrives
+ * pre-read in `icsByMessage` (see {@link resolveIcsByMessage}). Only updates,
* cancellations, and reply chains bundle; bare invites and RSVPs are skipped.
*/
export function classifyCalendarThread(
- messages: GmailMessage[]
+ messages: GmailMessage[],
+ icsByMessage: Map
): { uid: string; kind: "reply" | "update" | "cancel" } | null {
// 1. Reply chain — our header on any message.
for (const m of messages) {
@@ -807,7 +893,7 @@ export function classifyCalendarThread(
}
// 2. Calendar-system ICS.
for (const m of messages) {
- const ics = findPartContent(m.payload, "text/calendar");
+ const ics = icsByMessage.get(m.id);
if (!ics) continue;
const uid = icsProp(ics, "UID");
if (!uid) continue;
@@ -926,21 +1012,22 @@ function commentFromBody(message: GmailMessage): string | null {
/**
* Extract every attendee response carried by a Gmail conversation.
*
- * A descriptor is produced for each message whose payload holds a
- * `text/calendar` part with `METHOD:REPLY`, a `UID`, an `ATTENDEE` with a
- * decided `PARTSTAT`, and a resolvable attendee address. `NEEDS-ACTION`
- * yields nothing — there is no response to report.
+ * A descriptor is produced for each message whose calendar body — read ahead
+ * of time by {@link resolveIcsByMessage} — carries `METHOD:REPLY`, a `UID`, an
+ * `ATTENDEE` with a decided `PARTSTAT`, and a resolvable attendee address.
+ * `NEEDS-ACTION` yields nothing — there is no response to report.
*
* Every reply message is returned rather than only the first, so a
* conversation carrying a revised response stays correct.
*/
export function extractCalendarReplies(
- messages: GmailMessage[]
+ messages: GmailMessage[],
+ icsByMessage: Map
): CalendarReply[] {
const replies: CalendarReply[] = [];
for (const message of messages) {
- const ics = findPartContent(message.payload, "text/calendar");
+ const ics = icsByMessage.get(message.id);
if (!ics) continue;
if ((icsProp(ics, "METHOD") ?? "").toUpperCase() !== "REPLY") continue;
diff --git a/connectors/google/src/mail/sync.test.ts b/connectors/google/src/mail/sync.test.ts
index 4b56f29a..de1a38f3 100644
--- a/connectors/google/src/mail/sync.test.ts
+++ b/connectors/google/src/mail/sync.test.ts
@@ -16,6 +16,7 @@ import {
onCreateLinkFn,
onNoteCreatedFn,
onNoteReactionChangedFn,
+ drainPendingRsvpsFn,
processEmailThreadsFn,
REACTION_SEND_DELAY_MS,
sendReactionEmailFn,
@@ -91,6 +92,7 @@ function makeHost(): { host: GmailSyncHost; store: Map } {
saveNote: vi.fn(async () => null),
channelSyncCompleted: vi.fn(async () => {}),
setThreadToDo: vi.fn(async () => {}),
+ archiveLinks: vi.fn(async () => {}),
},
files: { read: vi.fn() },
network: { createWebhook: vi.fn(), deleteWebhook: vi.fn() },
@@ -129,7 +131,11 @@ function forwardDraft(overrides: Partial = {}): CreateLinkDraft
} as CreateLinkDraft;
}
-afterEach(() => vi.restoreAllMocks());
+afterEach(() => {
+ vi.restoreAllMocks();
+ vi.unstubAllGlobals();
+ registeredIcs.clear();
+});
describe("onCreateLinkFn — draft.forward", () => {
it("builds and sends a native Gmail forward of the source message", async () => {
@@ -815,7 +821,48 @@ function part(
};
}
-/** A single-message GmailThread carrying a `text/calendar` ICS part. */
+/**
+ * ICS bodies keyed by the attachment id Gmail would hand out for them.
+ *
+ * Gmail never leaves a calendar part inline: it synthesizes a filename
+ * (`invite.ics`) and moves the body out to `attachmentId`, so reading the ICS
+ * takes a second `messages.attachments.get` call. Fixtures below build that
+ * real shape and register the body here, and {@link serveIcsAttachments}
+ * stubs `fetch` to return it — so these tests exercise the same two-step read
+ * production performs rather than a payload shape Gmail never produces.
+ */
+const registeredIcs = new Map();
+
+/** Stubs `fetch` so `messages.attachments.get` serves the registered bodies. */
+function serveIcsAttachments(): void {
+ vi.stubGlobal(
+ "fetch",
+ vi.fn(async (url: string) => {
+ const attachmentId = url.split("/attachments/")[1]?.split("?")[0];
+ const ics = attachmentId ? registeredIcs.get(attachmentId) : undefined;
+ if (!ics) return new Response(null, { status: 404 });
+ return new Response(
+ JSON.stringify({ data: b64url(ics), size: ics.length }),
+ { status: 200, headers: { "Content-Type": "application/json" } }
+ );
+ })
+ );
+}
+
+/** A calendar part shaped the way Gmail really delivers one. */
+function icsAttachmentPart(ics: string): GmailMessagePart {
+ const attachmentId = `att-${registeredIcs.size + 1}`;
+ registeredIcs.set(attachmentId, ics);
+ serveIcsAttachments();
+ return {
+ mimeType: "text/calendar",
+ filename: "invite.ics",
+ headers: [],
+ body: { size: ics.length, attachmentId },
+ };
+}
+
+/** A single-message GmailThread carrying a calendar ICS part. */
function calendarUpdateThread(threadId: string, ics: string): GmailThread {
const message: GmailMessage = {
id: `${threadId}-msg-1`,
@@ -833,7 +880,7 @@ function calendarUpdateThread(threadId: string, ics: string): GmailThread {
],
parts: [
part("text/plain", { data: "The event has been updated." }),
- part("text/calendar", { data: ics }),
+ icsAttachmentPart(ics),
],
}),
};
@@ -928,7 +975,7 @@ function rsvpThread(
],
parts: [
part("text/plain", { data: "Beth Round has declined this invitation." }),
- part("text/calendar", { data: ics }),
+ icsAttachmentPart(ics),
],
}),
};
@@ -1101,6 +1148,148 @@ describe("processEmailThreadsFn — attendee responses fold onto the event", ()
);
expect(keys).toEqual(["rsvp-orphan-msg-1"]);
});
+
+ it("records a pending retry when the event has not synced yet", async () => {
+ const { host, store } = makeHost();
+ captureSaves(host, { noteId: null });
+
+ await processEmailThreadsFn(
+ host,
+ [rsvpThread("rsvp-pending", replyIcs("DECLINED"))],
+ false,
+ "INBOX"
+ );
+
+ expect(store.get("pending-rsvp:rsvp-pending")).toMatchObject({
+ threadId: "rsvp-pending",
+ });
+ });
+
+ it("does not track a conversation that also carries real correspondence", async () => {
+ const { host, store } = makeHost();
+ captureSaves(host, { noteId: null });
+
+ await processEmailThreadsFn(
+ host,
+ [
+ rsvpThread("rsvp-mixed", replyIcs("DECLINED"), {
+ withPlainReply: true,
+ }),
+ ],
+ false,
+ "INBOX"
+ );
+
+ // Archiving later must never hide a human reply, so a mixed conversation
+ // is left alone entirely.
+ expect(store.get("pending-rsvp:rsvp-mixed")).toBeUndefined();
+ });
+});
+
+describe("drainPendingRsvpsFn — retract once the event arrives", () => {
+ /** Seeds one pending entry and the Gmail thread the retry re-reads. */
+ function seedPending(
+ host: GmailSyncHost,
+ store: Map,
+ opts: { firstSeen?: string } = {}
+ ) {
+ const gmailThread = rsvpThread("rsvp-late", replyIcs("ACCEPTED"));
+ store.set("pending-rsvp:rsvp-late", {
+ threadId: "rsvp-late",
+ channelId: "INBOX",
+ firstSeen: opts.firstSeen ?? new Date().toISOString(),
+ });
+ (host.tools.store.list as ReturnType).mockImplementation(
+ async (prefix: string) =>
+ [...store.keys()].filter((k) => k.startsWith(prefix))
+ );
+ vi.spyOn(GmailApi.prototype, "getThread").mockResolvedValue(gmailThread);
+ }
+
+ it("folds the response and archives the standalone email thread", async () => {
+ const { host, store } = makeHost();
+ const { notes } = captureSaves(host, { noteId: "N" });
+ seedPending(host, store);
+
+ await drainPendingRsvpsFn(host);
+
+ expect(notes).toHaveLength(1);
+ expect(notes[0]).toMatchObject({
+ thread: { source: "icaluid:uid-rsvp@google.com" },
+ });
+ expect(host.tools.integrations.archiveLinks).toHaveBeenCalledWith({
+ meta: { threadId: "rsvp-late" },
+ });
+ expect(store.has("pending-rsvp:rsvp-late")).toBe(false);
+ });
+
+ it("applies the same unread rule the first pass would have", async () => {
+ const { host, store } = makeHost();
+ const { notes } = captureSaves(host, { noteId: "N" });
+ const declined = rsvpThread("rsvp-late", replyIcs("DECLINED"));
+ store.set("pending-rsvp:rsvp-late", {
+ threadId: "rsvp-late",
+ channelId: "INBOX",
+ initialSync: false,
+ firstSeen: new Date().toISOString(),
+ });
+ (host.tools.store.list as ReturnType).mockImplementation(
+ async (prefix: string) =>
+ [...store.keys()].filter((k) => k.startsWith(prefix))
+ );
+ vi.spyOn(GmailApi.prototype, "getThread").mockResolvedValue(declined);
+
+ await drainPendingRsvpsFn(host);
+
+ // A decline is worth surfacing; an acceptance is not. Retrying must not
+ // change that, or a late fold is noisier than a timely one.
+ expect(notes[0]).toMatchObject({ unread: true });
+ });
+
+ it("leaves a response first seen during the initial backfill read", async () => {
+ const { host, store } = makeHost();
+ const { notes } = captureSaves(host, { noteId: "N" });
+ const declined = rsvpThread("rsvp-late", replyIcs("DECLINED"));
+ store.set("pending-rsvp:rsvp-late", {
+ threadId: "rsvp-late",
+ channelId: "INBOX",
+ initialSync: true,
+ firstSeen: new Date().toISOString(),
+ });
+ (host.tools.store.list as ReturnType).mockImplementation(
+ async (prefix: string) =>
+ [...store.keys()].filter((k) => k.startsWith(prefix))
+ );
+ vi.spyOn(GmailApi.prototype, "getThread").mockResolvedValue(declined);
+
+ await drainPendingRsvpsFn(host);
+
+ expect(notes[0].unread).toBeUndefined();
+ });
+
+ it("keeps the entry and archives nothing while the event is still missing", async () => {
+ const { host, store } = makeHost();
+ captureSaves(host, { noteId: null });
+ seedPending(host, store);
+
+ await drainPendingRsvpsFn(host);
+
+ expect(host.tools.integrations.archiveLinks).not.toHaveBeenCalled();
+ expect(store.has("pending-rsvp:rsvp-late")).toBe(true);
+ });
+
+ it("gives up on an entry older than the retry window", async () => {
+ const { host, store } = makeHost();
+ captureSaves(host, { noteId: "N" });
+ seedPending(host, store, {
+ firstSeen: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000).toISOString(),
+ });
+
+ await drainPendingRsvpsFn(host);
+
+ expect(host.tools.integrations.archiveLinks).not.toHaveBeenCalled();
+ expect(store.has("pending-rsvp:rsvp-late")).toBe(false);
+ });
});
/** A single-message GmailThread carrying `labels`, with a plain-text body. */
diff --git a/connectors/google/src/mail/sync.ts b/connectors/google/src/mail/sync.ts
index a1de8e4d..cee65cd7 100644
--- a/connectors/google/src/mail/sync.ts
+++ b/connectors/google/src/mail/sync.ts
@@ -55,10 +55,12 @@ import {
extractCalendarReplies,
formatFromHeader,
getHeader,
+ hasCalendarPart,
isGmailRateLimitError,
isSendableGmailReaction,
mapWithConcurrency,
parseEmailAddresses,
+ resolveIcsByMessage,
syncGmailChannel,
syncGmailMailboxIncremental,
transformGmailThread,
@@ -307,6 +309,16 @@ export interface GmailSyncHost {
actorId: ActorId,
todo: boolean
): Promise;
+ /**
+ * Archive this connector's links matching a filter. A thread whose last
+ * non-archived link is archived is archived too.
+ */
+ archiveLinks(filter: {
+ channelId?: string;
+ type?: string;
+ status?: string;
+ meta?: Record;
+ }): Promise;
};
files: {
/** Read a file referenced by a note action (for outbound attachments). */
@@ -1469,9 +1481,146 @@ export async function processEmailThreadsFn(
// per-call save fan-out. Threads are independent — all per-thread state
// keys (`sent:`, `unread:`, `starred:`) are keyed by thread/message id —
// so ordering across threads carries no meaning.
+ // Calendar bodies for the whole batch. Gmail stores every calendar part as
+ // a separate attachment, so reading one costs an extra API call — done once
+ // here, ahead of the save fan-out, and skipped entirely for the
+ // overwhelmingly common batch that carries no calendar mail at all.
+ const icsByMessage = await resolveBatchIcsFn(
+ host,
+ transformed.flatMap(({ thread }) => thread.messages ?? [])
+ );
+
await mapWithConcurrency(transformed, SAVE_CONCURRENCY, (item) =>
- saveTransformedThread(host, item, initialSync)
+ saveTransformedThread(host, item, initialSync, icsByMessage)
);
+
+ // Responses whose event had not synced when they arrived. Retried after the
+ // saves above so an event that landed in this very batch is already there.
+ await drainPendingRsvpsFn(host);
+}
+
+/** Storage prefix for responses awaiting their event. */
+const PENDING_RSVP_PREFIX = "pending-rsvp:";
+
+/**
+ * How long a response keeps being retried before we stop. Past this the email
+ * thread simply stays as it is — the response is still readable, just not
+ * folded onto an event that never arrived.
+ */
+const PENDING_RSVP_TTL_MS = 7 * 24 * 60 * 60 * 1000;
+
+/** A response whose event thread had not synced when it first arrived. */
+type PendingRsvp = {
+ /**
+ * Gmail thread id. The retry re-fetches and re-parses the conversation
+ * rather than storing the parsed response, so it always reflects the
+ * current state of the mail and shares one code path with first-pass sync.
+ */
+ threadId: string;
+ channelId: string;
+ /**
+ * Whether the response was first seen during the initial backfill. Carried
+ * so the retry applies the same unread rule the first pass would have — a
+ * late fold must not be noisier than a timely one.
+ */
+ initialSync: boolean;
+ /** ISO timestamp of the first failed fold, for {@link PENDING_RSVP_TTL_MS}. */
+ firstSeen: string;
+};
+
+/**
+ * Retries responses that could not fold because their event had not synced yet.
+ *
+ * The email thread was already saved when the fold first failed — dropping it
+ * would have lost the response outright — so a successful retry has to retract
+ * it: the note moves onto the event and the now-empty email thread is archived.
+ * Only conversations that were nothing but responses are ever tracked (see
+ * {@link saveTransformedThread}), so this can never archive real correspondence.
+ *
+ * Costs one storage list per pass and nothing else when there is nothing
+ * pending, which is the overwhelmingly common case.
+ */
+export async function drainPendingRsvpsFn(host: GmailSyncHost): Promise {
+ const keys = await host.tools.store.list(PENDING_RSVP_PREFIX);
+ if (keys.length === 0) return;
+
+ const api = await getApiAnyFn(host);
+ if (!api) return;
+
+ for (const key of keys) {
+ const pending = await host.get(key);
+ if (!pending) {
+ await host.clear(key);
+ continue;
+ }
+
+ if (Date.now() - new Date(pending.firstSeen).getTime() > PENDING_RSVP_TTL_MS) {
+ await host.clear(key);
+ continue;
+ }
+
+ try {
+ const thread = await api.getThread(pending.threadId);
+ const messages = thread.messages ?? [];
+ const icsByMessage = await resolveIcsByMessage(api, messages);
+ const replies = extractCalendarReplies(messages, icsByMessage);
+ if (replies.length === 0) {
+ // No longer a response conversation (message deleted, or the calendar
+ // part became unreadable). Nothing left to retry.
+ await host.clear(key);
+ continue;
+ }
+
+ let allFolded = true;
+ for (const reply of replies) {
+ const noteId = await host.tools.integrations.saveNote({
+ thread: { source: `icaluid:${reply.uid}` },
+ key: reply.messageId,
+ content: composeRsvpNote(reply),
+ contentType: "markdown",
+ created: reply.sourceCreatedAt,
+ author: {
+ email: reply.attendeeEmail,
+ ...(reply.attendeeName ? { name: reply.attendeeName } : {}),
+ },
+ ...(shouldMarkUnread(reply, pending.initialSync)
+ ? { unread: true }
+ : {}),
+ });
+ if (!noteId) allFolded = false;
+ }
+ // Retract only once every response reached the event, so a partially
+ // folded conversation is never left with nowhere to read the rest.
+ if (!allFolded) continue;
+
+ await host.tools.integrations.archiveLinks({
+ meta: { threadId: pending.threadId },
+ });
+ await host.clear(key);
+ } catch (error) {
+ // Leave the entry in place; the next pass retries it.
+ console.warn(
+ `[gmail] could not retry the folded response for thread ${pending.threadId}:`,
+ error
+ );
+ }
+ }
+}
+
+/**
+ * Reads the iCalendar bodies for a batch of messages. Resolving an API client
+ * is itself a token round-trip, so a batch with no calendar part at all skips
+ * straight to an empty map. A batch whose connection has no usable token
+ * degrades the same way: the conversations sync as plain email.
+ */
+async function resolveBatchIcsFn(
+ host: GmailSyncHost,
+ messages: GmailMessage[]
+): Promise> {
+ if (!messages.some(hasCalendarPart)) return new Map();
+ const api = await getApiAnyFn(host);
+ if (!api) return new Map();
+ return resolveIcsByMessage(api, messages);
}
/**
@@ -1483,7 +1632,8 @@ export async function processEmailThreadsFn(
async function saveTransformedThread(
host: GmailSyncHost,
{ thread, plot: plotThread, channelId }: TransformedGmailThread,
- initialSync: boolean
+ initialSync: boolean,
+ icsByMessage: Map
): Promise {
try {
if (!plotThread.notes || plotThread.notes.length === 0) return;
@@ -1516,7 +1666,7 @@ async function saveTransformedThread(
// note was folded away — otherwise a mixed conversation gets its preview
// and classification from an RSVP notification that no longer has a note.
const foldedMessageIds = new Set();
- const replies = extractCalendarReplies(thread.messages ?? []);
+ const replies = extractCalendarReplies(thread.messages ?? [], icsByMessage);
if (replies.length > 0) {
for (const reply of replies) {
// A miss means the calendar event has not synced yet (saveNote returns
@@ -1537,6 +1687,29 @@ async function saveTransformedThread(
});
if (noteId) foldedMessageIds.add(reply.messageId);
}
+
+ // Any response that missed its event is worth retrying: the calendar
+ // often syncs seconds later. Only track a conversation that is nothing
+ // but responses — the retry retracts the email thread by archiving it,
+ // which must never hide a human reply.
+ const unfolded = replies.filter((r) => !foldedMessageIds.has(r.messageId));
+ const replyMessageIds = new Set(replies.map((r) => r.messageId));
+ const isResponsesOnly = plotThread.notes.every((note) => {
+ const noteKey = "key" in note ? (note as { key: string }).key : null;
+ return noteKey !== null && replyMessageIds.has(noteKey);
+ });
+ if (unfolded.length > 0 && isResponsesOnly) {
+ const key = `${PENDING_RSVP_PREFIX}${thread.id}`;
+ const existing = await host.get(key);
+ await host.set(key, {
+ threadId: thread.id,
+ channelId,
+ initialSync,
+ // Preserved across passes so the retry window measures from the
+ // first failure, not from the most recent re-sync of the thread.
+ firstSeen: existing?.firstSeen ?? new Date().toISOString(),
+ } satisfies PendingRsvp);
+ }
if (foldedMessageIds.size > 0) {
plotThread.notes = plotThread.notes.filter((note) => {
const noteKey = "key" in note ? (note as { key: string }).key : null;
@@ -1601,7 +1774,10 @@ async function saveTransformedThread(
// Bundle onto the calendar event's thread when this conversation relates
// to one (a Plot-sent reply chain, or an ICS update/cancellation).
- const calBundle = classifyCalendarThread(thread.messages ?? []);
+ const calBundle = classifyCalendarThread(
+ thread.messages ?? [],
+ icsByMessage
+ );
if (calBundle) {
plotThread.sources = [
...(plotThread.sources ?? []),