From 81840b3cc23481e230bd41942be4b514f3815e88 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Mon, 3 Aug 2026 13:45:08 -0400 Subject: [PATCH 1/4] refactor: share the bounded-concurrency helper across product modules Moves mapWithConcurrency out of the Gmail mail module into a new connectors/google/src/concurrency.ts, re-exported from gmail-api.ts for existing callers. This lets other product modules within the Google connector (e.g. calendar) reuse the same bounded-concurrency helper without importing from the mail module. --- connectors/google/src/concurrency.test.ts | 30 +++++++++++++++++++++++ connectors/google/src/concurrency.ts | 27 ++++++++++++++++++++ connectors/google/src/mail/gmail-api.ts | 27 +++----------------- 3 files changed, 60 insertions(+), 24 deletions(-) create mode 100644 connectors/google/src/concurrency.test.ts create mode 100644 connectors/google/src/concurrency.ts diff --git a/connectors/google/src/concurrency.test.ts b/connectors/google/src/concurrency.test.ts new file mode 100644 index 00000000..ce14e308 --- /dev/null +++ b/connectors/google/src/concurrency.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, it } from "vitest"; +import { mapWithConcurrency } from "./concurrency"; + +describe("mapWithConcurrency", () => { + it("returns results in input order regardless of completion order", async () => { + const delays = [30, 0, 15]; + const result = await mapWithConcurrency(delays, 3, async (ms, i) => { + await new Promise((r) => setTimeout(r, ms)); + return i; + }); + expect(result).toEqual([0, 1, 2]); + }); + + it("never runs more than `limit` mappers at once", async () => { + let inFlight = 0; + let peak = 0; + await mapWithConcurrency([1, 2, 3, 4, 5, 6, 7], 3, async () => { + inFlight++; + peak = Math.max(peak, inFlight); + await new Promise((r) => setTimeout(r, 5)); + inFlight--; + return null; + }); + expect(peak).toBe(3); + }); + + it("resolves to an empty array for empty input", async () => { + expect(await mapWithConcurrency([], 5, async () => 1)).toEqual([]); + }); +}); diff --git a/connectors/google/src/concurrency.ts b/connectors/google/src/concurrency.ts new file mode 100644 index 00000000..0b4c7ca0 --- /dev/null +++ b/connectors/google/src/concurrency.ts @@ -0,0 +1,27 @@ +/** + * Maps `items` with at most `limit` mappers in flight, resolving to results in + * input order. A mapper rejection rejects the whole call — callers that want + * per-item failure semantics catch inside `fn`. + * + * Shared by the mail and calendar product modules. It lives here rather than in + * either module so neither has to import from the other. + */ +export async function mapWithConcurrency( + items: readonly T[], + limit: number, + fn: (item: T, index: number) => Promise +): Promise { + const results = new Array(items.length); + let next = 0; + const workers = Array.from( + { length: Math.min(limit, items.length) }, + async () => { + while (next < items.length) { + const index = next++; + results[index] = await fn(items[index], index); + } + } + ); + await Promise.all(workers); + return results; +} diff --git a/connectors/google/src/mail/gmail-api.ts b/connectors/google/src/mail/gmail-api.ts index 0a865fe5..428bd1ec 100644 --- a/connectors/google/src/mail/gmail-api.ts +++ b/connectors/google/src/mail/gmail-api.ts @@ -10,6 +10,7 @@ import type { import { markdownToPlainText } from "@plotday/twister/utils/markdown"; import { markdownToHtml } from "@plotday/twister/utils/markdown-html"; import { isNoReplySender } from "@plotday/twister/signals"; +import { mapWithConcurrency } from "../concurrency"; export type GmailLabel = { @@ -1644,30 +1645,8 @@ async function syncGmailChannelIncremental( */ const THREAD_FETCH_CONCURRENCY = 5; -/** - * Maps `items` with at most `limit` mappers in flight, resolving to results in - * input order. A mapper rejection rejects the whole call — callers that want - * per-item failure semantics catch inside `fn`. - */ -export async function mapWithConcurrency( - items: readonly T[], - limit: number, - fn: (item: T, index: number) => Promise -): Promise { - const results = new Array(items.length); - let next = 0; - const workers = Array.from( - { length: Math.min(limit, items.length) }, - async () => { - while (next < items.length) { - const index = next++; - results[index] = await fn(items[index], index); - } - } - ); - await Promise.all(workers); - return results; -} +// Re-exported for existing callers (mail/sync.ts imports it from here). +export { mapWithConcurrency }; /** * Builds the Gmail search bound for a backfill cursor: the channel's own query From 0f51a10c737e10847f1c321b495ac32912197256 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Mon, 3 Aug 2026 13:51:09 -0400 Subject: [PATCH 2/4] feat(calendar): bound imported history to 366 days --- connectors/google/src/calendar/sync.test.ts | 16 ++++++++++++++++ connectors/google/src/calendar/sync.ts | 16 +++++++--------- 2 files changed, 23 insertions(+), 9 deletions(-) diff --git a/connectors/google/src/calendar/sync.test.ts b/connectors/google/src/calendar/sync.test.ts index 06c10b10..1f47a6d5 100644 --- a/connectors/google/src/calendar/sync.test.ts +++ b/connectors/google/src/calendar/sync.test.ts @@ -11,6 +11,7 @@ import type { Thread } from "@plotday/twister"; import type { CalendarSyncHost } from "./sync"; import { buildEventSources, + calendarHistoryFloor, cancelEventWithApiFn, cancellationWasSelfInitiatedFn, extractRSVPParamsFn, @@ -2148,3 +2149,18 @@ describe("processCalendarEventsFn — initial-sync occurrence round-trips", () = expect(totalRoundTrips(large.calls)).toBeLessThan(10); }); }); + +describe("calendarHistoryFloor", () => { + it("is exactly 366 days before now, so annual events stay in window", () => { + const now = new Date("2026-08-03T12:00:00.000Z"); + const floor = calendarHistoryFloor(now); + const days = (now.getTime() - floor.getTime()) / (24 * 60 * 60 * 1000); + expect(days).toBe(366); + }); + + it("keeps an event dated exactly one year ago inside the window", () => { + const now = new Date("2026-08-03T12:00:00.000Z"); + const oneYearAgo = new Date("2025-08-03T12:00:00.000Z"); + expect(oneYearAgo >= calendarHistoryFloor(now)).toBe(true); + }); +}); diff --git a/connectors/google/src/calendar/sync.ts b/connectors/google/src/calendar/sync.ts index 4d04d5d9..622674c1 100644 --- a/connectors/google/src/calendar/sync.ts +++ b/connectors/google/src/calendar/sync.ts @@ -283,17 +283,15 @@ export async function firstSeenAtFn( } /** - * Start of the history window the initial backfill imports: Jan 1 of two - * calendar years ago (mirrors the `historyMin` computed when the quick pass - * transitions to the full pass). Events scheduled before this were never - * imported, so a cancellation for one can only materialise a phantom thread. + * How far back the initial backfill imports: 366 days — a year and a day, so an + * annual event is always inside the window. Events scheduled before this were + * never imported, so a cancellation for one can only materialise a phantom + * thread (see `cancellationIsForUnimportedEventFn`). */ +export const CALENDAR_HISTORY_DAYS = 366; + export function calendarHistoryFloor(now: Date = new Date()): Date { - const floor = new Date(now); - floor.setFullYear(floor.getFullYear() - 2); - floor.setMonth(0, 1); - floor.setHours(0, 0, 0, 0); - return floor; + return new Date(now.getTime() - CALENDAR_HISTORY_DAYS * 24 * 60 * 60 * 1000); } /** From 765838f7480973bbff6363c107a6993fa71a9fdc Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Mon, 3 Aug 2026 14:00:00 -0400 Subject: [PATCH 3/4] Revert "refactor: share the bounded-concurrency helper across product modules" This reverts commit 81840b3cc23481e230bd41942be4b514f3815e88. --- connectors/google/src/concurrency.test.ts | 30 ----------------------- connectors/google/src/concurrency.ts | 27 -------------------- connectors/google/src/mail/gmail-api.ts | 27 +++++++++++++++++--- 3 files changed, 24 insertions(+), 60 deletions(-) delete mode 100644 connectors/google/src/concurrency.test.ts delete mode 100644 connectors/google/src/concurrency.ts diff --git a/connectors/google/src/concurrency.test.ts b/connectors/google/src/concurrency.test.ts deleted file mode 100644 index ce14e308..00000000 --- a/connectors/google/src/concurrency.test.ts +++ /dev/null @@ -1,30 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { mapWithConcurrency } from "./concurrency"; - -describe("mapWithConcurrency", () => { - it("returns results in input order regardless of completion order", async () => { - const delays = [30, 0, 15]; - const result = await mapWithConcurrency(delays, 3, async (ms, i) => { - await new Promise((r) => setTimeout(r, ms)); - return i; - }); - expect(result).toEqual([0, 1, 2]); - }); - - it("never runs more than `limit` mappers at once", async () => { - let inFlight = 0; - let peak = 0; - await mapWithConcurrency([1, 2, 3, 4, 5, 6, 7], 3, async () => { - inFlight++; - peak = Math.max(peak, inFlight); - await new Promise((r) => setTimeout(r, 5)); - inFlight--; - return null; - }); - expect(peak).toBe(3); - }); - - it("resolves to an empty array for empty input", async () => { - expect(await mapWithConcurrency([], 5, async () => 1)).toEqual([]); - }); -}); diff --git a/connectors/google/src/concurrency.ts b/connectors/google/src/concurrency.ts deleted file mode 100644 index 0b4c7ca0..00000000 --- a/connectors/google/src/concurrency.ts +++ /dev/null @@ -1,27 +0,0 @@ -/** - * Maps `items` with at most `limit` mappers in flight, resolving to results in - * input order. A mapper rejection rejects the whole call — callers that want - * per-item failure semantics catch inside `fn`. - * - * Shared by the mail and calendar product modules. It lives here rather than in - * either module so neither has to import from the other. - */ -export async function mapWithConcurrency( - items: readonly T[], - limit: number, - fn: (item: T, index: number) => Promise -): Promise { - const results = new Array(items.length); - let next = 0; - const workers = Array.from( - { length: Math.min(limit, items.length) }, - async () => { - while (next < items.length) { - const index = next++; - results[index] = await fn(items[index], index); - } - } - ); - await Promise.all(workers); - return results; -} diff --git a/connectors/google/src/mail/gmail-api.ts b/connectors/google/src/mail/gmail-api.ts index 428bd1ec..0a865fe5 100644 --- a/connectors/google/src/mail/gmail-api.ts +++ b/connectors/google/src/mail/gmail-api.ts @@ -10,7 +10,6 @@ import type { import { markdownToPlainText } from "@plotday/twister/utils/markdown"; import { markdownToHtml } from "@plotday/twister/utils/markdown-html"; import { isNoReplySender } from "@plotday/twister/signals"; -import { mapWithConcurrency } from "../concurrency"; export type GmailLabel = { @@ -1645,8 +1644,30 @@ async function syncGmailChannelIncremental( */ const THREAD_FETCH_CONCURRENCY = 5; -// Re-exported for existing callers (mail/sync.ts imports it from here). -export { mapWithConcurrency }; +/** + * Maps `items` with at most `limit` mappers in flight, resolving to results in + * input order. A mapper rejection rejects the whole call — callers that want + * per-item failure semantics catch inside `fn`. + */ +export async function mapWithConcurrency( + items: readonly T[], + limit: number, + fn: (item: T, index: number) => Promise +): Promise { + const results = new Array(items.length); + let next = 0; + const workers = Array.from( + { length: Math.min(limit, items.length) }, + async () => { + while (next < items.length) { + const index = next++; + results[index] = await fn(items[index], index); + } + } + ); + await Promise.all(workers); + return results; +} /** * Builds the Gmail search bound for a backfill cursor: the channel's own query From 5e7375ede5b88d8a2209f2de9a2aaa9e5c915fd3 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Mon, 3 Aug 2026 14:00:36 -0400 Subject: [PATCH 4/4] docs(calendar): correct what the Google quick pass actually returns The comment on SyncState.phase claimed the quick pass front-loads only non-recurring events and that the full pass is what picks up long-running recurring masters "excluded by the quick pass's timeMin". That is wrong. Google does not apply timeMin to recurring masters: verified against a live calendar, a FREQ=YEARLY master whose first instance was in 1980 and a FREQ=WEEKLY master whose RRULE UNTIL had already passed were both returned by a timeMin=now listing. Only non-recurring events are bounded by timeMin, so the quick pass already surfaces upcoming recurring meetings and the full pass exists for past one-off events and the nextSyncToken. Also notes that this is Google-specific and must not be generalised to Microsoft Graph, whose $filter on start/dateTime is a literal comparison against the series master's own first start. --- connectors/google/src/calendar/google-api.ts | 24 ++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/connectors/google/src/calendar/google-api.ts b/connectors/google/src/calendar/google-api.ts index 1284db5d..cc85b0f0 100644 --- a/connectors/google/src/calendar/google-api.ts +++ b/connectors/google/src/calendar/google-api.ts @@ -91,10 +91,26 @@ export type SyncState = { min?: Date | null; max?: Date | null; sequence?: number; - // Initial sync runs two passes: "quick" (timeMin=now, front-loads upcoming - // non-recurring events and future exceptions) then "full" (timeMin=history - // limit, picks up long-running recurring masters excluded by the quick - // pass's timeMin). Undefined on incremental webhook-triggered syncs. + // Initial sync runs two passes: "quick" (timeMin=now) then "full" + // (timeMin=history floor). + // + // The quick pass front-loads everything upcoming, recurring series + // INCLUDED: Google does not apply `timeMin` to recurring masters. Verified + // against a live calendar — a FREQ=YEARLY master whose first instance was + // in 1980, and a FREQ=WEEKLY master whose RRULE `UNTIL` had already passed, + // were both returned by a `timeMin=now` listing. Only non-recurring events + // are actually bounded by `timeMin`. + // + // So the full pass is NOT what makes recurring meetings appear (an earlier + // version of this comment claimed it was). It exists to import past one-off + // events and to establish the `nextSyncToken` that incremental syncs reuse. + // + // Do not generalise this to Outlook: Microsoft Graph's + // `$filter=start/dateTime ge …` is a literal comparison against the series + // master's own (first) start, which genuinely does exclude a long-running + // master. See `outlook/src/calendar/sync.ts`. + // + // Undefined on incremental webhook-triggered syncs. phase?: "quick" | "full"; };