diff --git a/apps/convex/__tests__/googleSyncLoop.test.ts b/apps/convex/__tests__/googleSyncLoop.test.ts new file mode 100644 index 000000000..d0dde7724 --- /dev/null +++ b/apps/convex/__tests__/googleSyncLoop.test.ts @@ -0,0 +1,1595 @@ +/** + * THE FORWARD SYNC LOOP: who it polls, how often, and what it refuses. + * + * Every control in `functions/googleSync.ts` could be deleted with the rest of + * this suite green unless a test names the sabotage it catches, so each + * describe block below is one such name: + * + * - the sweep starts a pass only for connections that are **due**, and never + * for a disconnected one, one with nothing this engine can sync, or one + * whose pass is still running; + * - the five-minute floor is refused **server-side**, not merely absent from + * a picker; + * - only the owner of a personal context may change the interval, and an + * owner of a *different* context cannot tell "not yours" from "no such + * connection" — proved with attacker and victim in ONE database, because + * two databases would make the refusal come from the row not existing; + * - the cursor advances over mail that was written and **not** over mail that + * was not; + * - a connection that has never synced does not look like one syncing fine. + * + * The end-to-end block drives the real `runFileOperation` — the credential + * barrier — against a fixture Gmail and an in-memory S3, so the pass under + * test is the pass that ships, including `gmailSync.js`'s own rendering. + * + * Every value here is obviously fake. This repository is public. + */ + +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { api, internal } from "../_generated/api"; +import type { Id } from "../_generated/dataModel"; +import { + addMember, + asUser, + captureError, + createUser, + createWorkspace, + errorCode, + seedGoogleConnection, + setupTest, + FAKE_STORAGE, + type TestConvex, +} from "./fixtures.helpers"; +import { memoryS3, type MemoryS3 } from "./storeStub.helpers"; +import { encryptSecret, requireKeyset } from "../functions/lib/crypto"; +import { + DEFAULT_SYNC_INTERVAL_MINUTES, + MAX_SYNC_BACKOFF_MS, + MAX_SYNC_INTERVAL_MINUTES, + MIN_SYNC_INTERVAL_MINUTES, + SYNC_FAILURE_BACKOFF_MS, + SYNC_STALL_MS, +} from "../functions/lib/googleSchedule"; + +const MINUTE = 60_000; + +function enableMailSync() { + vi.stubEnv("MAIL_CONNECT_ENABLED", "true"); + vi.stubEnv("GOOGLE_OAUTH_CLIENT_ID", "test-google-client-id.apps.googleusercontent.com"); +} + +afterEach(() => { + vi.unstubAllEnvs(); + vi.unstubAllGlobals(); +}); + +interface Scenario { + t: TestConvex; + owner: Id<"users">; + workspaceId: Id<"workspaces">; + connectionId: Id<"googleConnections">; +} + +async function scenario(): Promise { + const t = setupTest(); + const owner = await createUser(t, "owner@example.invalid"); + const workspaceId = await createWorkspace(t, owner, "atlas"); + const connectionId = await seedGoogleConnection(t, { workspaceId, boundBy: owner }); + return { t, owner, workspaceId, connectionId }; +} + +/** Patch the connection row directly — the states a sweep meets, without a pass to reach them. */ +async function patchConnection( + t: TestConvex, + connectionId: Id<"googleConnections">, + patch: Record, +): Promise { + await t.run((ctx) => ctx.db.patch(connectionId, patch)); +} + +async function readConnection(t: TestConvex, connectionId: Id<"googleConnections">) { + const row = await t.run((ctx) => ctx.db.get(connectionId)); + if (row === null) throw new Error("connection vanished"); + return row; +} + +async function sweep(t: TestConvex): Promise<{ started: number; examined: number }> { + return await t.mutation(internal.functions.googleSync.sweepDueGoogleSyncs, {}); +} + +/* -------------------------------------------------------------------------- */ + +describe("the sweep starts a pass only for a connection that is due", () => { + beforeEach(() => enableMailSync()); + + test("a connection that has never synced is due at once", async () => { + const { t, connectionId } = await scenario(); + expect(await sweep(t)).toEqual({ started: 1, examined: 1 }); + const row = await readConnection(t, connectionId); + expect(row.syncStartedAt).toBeTypeOf("number"); + // Claimed, so the next tick five minutes from now finds it not due even + // if this pass never reports. + expect(row.nextSyncAt).toBeGreaterThan(Date.now()); + }); + + test("...and one whose interval has not elapsed is left alone", async () => { + const { t, connectionId } = await scenario(); + const now = Date.now(); + await patchConnection(t, connectionId, { + syncIntervalMinutes: 15, + lastSyncAt: now - 5 * MINUTE, + nextSyncAt: now + 10 * MINUTE, + }); + expect((await sweep(t)).started).toBe(0); + expect((await readConnection(t, connectionId)).syncStartedAt).toBeUndefined(); + }); + + test("...and one whose interval HAS elapsed is started", async () => { + const { t, connectionId } = await scenario(); + const now = Date.now(); + await patchConnection(t, connectionId, { + syncIntervalMinutes: 15, + lastSyncAt: now - 16 * MINUTE, + nextSyncAt: now - MINUTE, + }); + expect((await sweep(t)).started).toBe(1); + }); + + test("the due rule is lastSyncAt plus the interval, not a stale nextSyncAt", async () => { + const { t, connectionId } = await scenario(); + const now = Date.now(); + // `nextSyncAt` says due; the two facts it is computed from say otherwise — + // a row left behind by a longer interval being written to it. The + // materialized copy must not be able to out-vote them. + await patchConnection(t, connectionId, { + syncIntervalMinutes: 60, + lastSyncAt: now - 10 * MINUTE, + nextSyncAt: now - MINUTE, + }); + expect((await sweep(t)).started).toBe(0); + }); + + test("a disconnected connection is never started, however overdue it looks", async () => { + const { t, connectionId } = await scenario(); + const now = Date.now(); + await patchConnection(t, connectionId, { + disconnectedAt: now, + lastSyncAt: now - 10 * 60 * MINUTE, + nextSyncAt: now - 10 * 60 * MINUTE, + }); + expect(await sweep(t)).toEqual({ started: 0, examined: 0 }); + expect((await readConnection(t, connectionId)).syncStartedAt).toBeUndefined(); + }); + + test("a connection with no product this engine can advance is not started", async () => { + const { t, connectionId } = await scenario(); + await patchConnection(t, connectionId, { products: ["calendar"] }); + expect((await sweep(t)).started).toBe(0); + }); + + test("a pass that is still running is not overtaken", async () => { + const { t, connectionId } = await scenario(); + const now = Date.now(); + await patchConnection(t, connectionId, { + lastSyncAt: now - 60 * MINUTE, + nextSyncAt: now - 30 * MINUTE, + syncStartedAt: now - MINUTE, + }); + expect((await sweep(t)).started).toBe(0); + }); + + test("...but one that has been silent for fifteen minutes is presumed lost and restarted", async () => { + const { t, connectionId } = await scenario(); + const now = Date.now(); + await patchConnection(t, connectionId, { + lastSyncAt: now - 60 * MINUTE, + nextSyncAt: now - 30 * MINUTE, + syncStartedAt: now - SYNC_STALL_MS - MINUTE, + }); + expect((await sweep(t)).started).toBe(1); + }); + + test("a deployment where Google mail is not enabled sweeps nothing", async () => { + const { t, connectionId } = await scenario(); + vi.stubEnv("MAIL_CONNECT_ENABLED", ""); + expect(await sweep(t)).toEqual({ started: 0, examined: 0 }); + expect((await readConnection(t, connectionId)).syncStartedAt).toBeUndefined(); + }); + + test("one sweep claims a bounded batch, and the rest drain on later runs", async () => { + const t = setupTest(); + const owner = await createUser(t, "owner@example.invalid"); + const workspaceId = await createWorkspace(t, owner, "atlas"); + for (let index = 0; index < 60; index += 1) { + await seedGoogleConnection(t, { + workspaceId, + boundBy: owner, + address: `person-${index}@example.invalid`, + }); + } + const first = await sweep(t); + expect(first.started).toBe(50); + // The claimed fifty are no longer due, so the second run reaches the rest. + expect((await sweep(t)).started).toBe(10); + }); +}); + +describe("the five-minute floor is enforced server-side", () => { + beforeEach(() => enableMailSync()); + + test("four minutes is refused", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + const error = await captureError(() => + asUser(t, owner).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId, + connectionId, + syncIntervalMinutes: 4, + }), + ); + expect(errorCode(error)).toBe("GOOGLE_SYNC_INTERVAL_TOO_SHORT"); + expect((await readConnection(t, connectionId)).syncIntervalMinutes).toBeUndefined(); + }); + + test("zero and a negative number are refused the same way", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + for (const minutes of [0, -5]) { + const error = await captureError(() => + asUser(t, owner).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId, + connectionId, + syncIntervalMinutes: minutes, + }), + ); + expect(errorCode(error)).toBe("GOOGLE_SYNC_INTERVAL_TOO_SHORT"); + } + }); + + test("a fractional interval is refused rather than rounded", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + const error = await captureError(() => + asUser(t, owner).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId, + connectionId, + syncIntervalMinutes: 5.5, + }), + ); + expect(errorCode(error)).toBe("GOOGLE_SYNC_INTERVAL_INVALID"); + }); + + test("longer than a day is refused, because Gmail expires a cursor in about a week", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + const error = await captureError(() => + asUser(t, owner).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId, + connectionId, + syncIntervalMinutes: MAX_SYNC_INTERVAL_MINUTES + 1, + }), + ); + expect(errorCode(error)).toBe("GOOGLE_SYNC_INTERVAL_TOO_LONG"); + }); + + test("the floor itself is accepted", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + await asUser(t, owner).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId, + connectionId, + syncIntervalMinutes: MIN_SYNC_INTERVAL_MINUTES, + }); + expect((await readConnection(t, connectionId)).syncIntervalMinutes).toBe( + MIN_SYNC_INTERVAL_MINUTES, + ); + }); + + test("a connection nobody has chosen an interval for is polled at the default", async () => { + const { t, connectionId } = await scenario(); + const now = Date.now(); + await patchConnection(t, connectionId, { + lastSyncAt: now - (DEFAULT_SYNC_INTERVAL_MINUTES - 2) * MINUTE, + nextSyncAt: now - MINUTE, + }); + expect((await sweep(t)).started).toBe(0); + await patchConnection(t, connectionId, { + lastSyncAt: now - (DEFAULT_SYNC_INTERVAL_MINUTES + 1) * MINUTE, + }); + expect((await sweep(t)).started).toBe(1); + }); + + test("shortening the interval brings the next pass forward without waiting out the old one", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + const lastSyncAt = Date.now() - 10 * MINUTE; + await patchConnection(t, connectionId, { + syncIntervalMinutes: 60, + lastSyncAt, + nextSyncAt: lastSyncAt + 60 * MINUTE, + }); + expect((await sweep(t)).started).toBe(0); + + await asUser(t, owner).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId, + connectionId, + syncIntervalMinutes: 5, + }); + expect((await readConnection(t, connectionId)).nextSyncAt).toBe(lastSyncAt + 5 * MINUTE); + expect((await sweep(t)).started).toBe(1); + }); +}); + +describe("only this context's owner may change how often it syncs", () => { + beforeEach(() => enableMailSync()); + + test("an editor of the owner's personal context cannot", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + const editor = await createUser(t, "editor@example.invalid"); + await addMember(t, workspaceId, editor, "editor", owner); + const error = await captureError(() => + asUser(t, editor).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId, + connectionId, + syncIntervalMinutes: 5, + }), + ); + expect(errorCode(error)).toBe("NOT_OWNER"); + expect((await readConnection(t, connectionId)).syncIntervalMinutes).toBeUndefined(); + }); + + test("a stranger with a context of their own cannot", async () => { + const { t, workspaceId, connectionId } = await scenario(); + const stranger = await createUser(t, "stranger@example.invalid"); + await createWorkspace(t, stranger, "elsewhere"); + const error = await captureError(() => + asUser(t, stranger).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId, + connectionId, + syncIntervalMinutes: 5, + }), + ); + expect(errorCode(error)).toBe("NOT_OWNER"); + }); + + /* + ATTACKER AND VICTIM IN ONE DATABASE. + + The attacker owns a real personal context of her own, so `requirePersonalOwner` + says yes for the workspace she names — and the connection she names is + somebody else's, in a workspace she has never been a member of. Two separate + databases would prove nothing here: the refusal would come from the row not + existing. + */ + test("an owner of another context cannot reach this connection, and learns nothing about it", async () => { + const { t, workspaceId: victimWorkspace, connectionId: victimConnection } = await scenario(); + const attacker = await createUser(t, "attacker@example.invalid"); + const attackerWorkspace = await createWorkspace(t, attacker, "attacker"); + + const throughOwnContext = await captureError(() => + asUser(t, attacker).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId: attackerWorkspace, + connectionId: victimConnection, + syncIntervalMinutes: MAX_SYNC_INTERVAL_MINUTES, + }), + ); + const throughVictimContext = await captureError(() => + asUser(t, attacker).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId: victimWorkspace, + connectionId: victimConnection, + syncIntervalMinutes: MAX_SYNC_INTERVAL_MINUTES, + }), + ); + + expect(errorCode(throughOwnContext)).toBe("GOOGLE_CONNECTION_NOT_FOUND"); + expect(errorCode(throughVictimContext)).toBe("NOT_OWNER"); + // Naming a connection id that exists must answer exactly as naming one + // that never did. + const missing = await t.run(async (ctx) => { + const id = await ctx.db.insert("googleConnections", { + workspaceId: attackerWorkspace, + provider: "google" as const, + address: "ghost@example.invalid", + encryptedRefreshToken: "", + scopes: [], + googleAccountId: "example-ghost-account", + products: ["gmail" as const], + health: "active" as const, + boundBy: attacker, + createdAt: Date.now(), + updatedAt: Date.now(), + }); + await ctx.db.delete(id); + return id; + }); + const throughMissing = await captureError(() => + asUser(t, attacker).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId: attackerWorkspace, + connectionId: missing, + syncIntervalMinutes: MAX_SYNC_INTERVAL_MINUTES, + }), + ); + expect(errorCode(throughMissing)).toBe(errorCode(throughOwnContext)); + expect((await readConnection(t, victimConnection)).syncIntervalMinutes).toBeUndefined(); + expect((await readConnection(t, victimConnection)).nextSyncAt).toBeUndefined(); + }); + + test("the owner of a SHARED context cannot set one, since a mailbox never lands in one", async () => { + const t = setupTest(); + const owner = await createUser(t, "owner@example.invalid"); + const shared = await createWorkspace(t, owner, "atlas-team", { kind: "shared" }); + const connectionId = await seedGoogleConnection(t, { workspaceId: shared, boundBy: owner }); + const error = await captureError(() => + asUser(t, owner).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId: shared, + connectionId, + syncIntervalMinutes: 5, + }), + ); + expect(errorCode(error)).toBe("NOT_OWNER"); + }); + + test("a disconnected account has no schedule to change", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + await patchConnection(t, connectionId, { disconnectedAt: Date.now() }); + const error = await captureError(() => + asUser(t, owner).mutation(api.functions.googleSync.updateGoogleSyncInterval, { + workspaceId, + connectionId, + syncIntervalMinutes: 5, + }), + ); + expect(errorCode(error)).toBe("GOOGLE_CONNECTION_NOT_FOUND"); + }); +}); + +describe("what a pass writes back onto the row", () => { + beforeEach(() => enableMailSync()); + + async function claimed(): Promise { + const s = await scenario(); + await patchConnection(s.t, s.connectionId, { syncStartedAt: Date.now(), syncIntervalMinutes: 30 }); + return s; + } + + test("a synced pass advances the cursor, releases the claim, and schedules the next one", async () => { + const { t, connectionId } = await claimed(); + const accepted = await t.mutation( + internal.functions.googleSync.recordGoogleForwardSyncPass, + { + connectionId, + status: "synced", + historyId: "2000", + daysTouched: 2, + bytesWritten: 4096, + }, + ); + expect(accepted).toEqual({ accepted: true }); + const row = await readConnection(t, connectionId); + expect(row.gmail?.historyId).toBe("2000"); + expect(row.gmail?.lastSyncedAt).toBeTypeOf("number"); + expect(row.syncStartedAt).toBeUndefined(); + expect(row.syncBytesWritten).toBe(4096); + expect(row.health).toBe("active"); + expect(row.nextSyncAt).toBe(row.lastSyncAt! + 30 * MINUTE); + }); + + test("a failed pass leaves the cursor alone, records the failure, and backs off", async () => { + const { t, connectionId } = await claimed(); + await patchConnection(t, connectionId, { + gmail: { ...(await readConnection(t, connectionId)).gmail!, historyId: "1000" }, + }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "failed", + // A cursor offered by a pass that did not finish. The mutation is where + // this is refused, not the call site: a failed pass that advanced the + // cursor would skip whatever it could not write, permanently and + // silently, which is the one defect in this design that loses mail. + historyId: "2000", + errorCode: "GOOGLE_RATE_LIMITED", + error: "Google rate-limited this mailbox.", + }); + const row = await readConnection(t, connectionId); + expect(row.gmail?.historyId).toBe("1000"); + expect(row.gmail?.lastSyncedAt).toBeUndefined(); + expect(row.health).toBe("error"); + expect(row.errorCode).toBe("GOOGLE_RATE_LIMITED"); + expect(row.lastSyncFailureCode).toBe("GOOGLE_RATE_LIMITED"); + expect(row.syncStartedAt).toBeUndefined(); + // Thirty-minute interval, fifteen-minute backoff floor: the longer wins, + // plus this connection's own spread, so a deployment's connections do not + // all wake in the same minute after an outage ends. + const wait = row.nextSyncAt! - row.lastSyncAt!; + expect(wait).toBeGreaterThanOrEqual(30 * MINUTE); + expect(wait).toBeLessThan(31 * MINUTE); + }); + + test("...and a five-minute interval still waits out the failure backoff", async () => { + const { t, connectionId } = await claimed(); + await patchConnection(t, connectionId, { syncIntervalMinutes: 5 }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "failed", + errorCode: "GOOGLE_UNAVAILABLE", + error: "Google did not answer reliably.", + }); + const row = await readConnection(t, connectionId); + const wait = row.nextSyncAt! - row.lastSyncAt!; + expect(wait).toBeGreaterThanOrEqual(SYNC_FAILURE_BACKOFF_MS); + expect(wait).toBeLessThan(SYNC_FAILURE_BACKOFF_MS + MINUTE); + }); + + test("the last failure survives a later success, because that is the question being asked", async () => { + const { t, connectionId } = await claimed(); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "failed", + errorCode: "GOOGLE_UNAVAILABLE", + error: "Google did not answer reliably.", + }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "synced", + historyId: "3000", + }); + const row = await readConnection(t, connectionId); + expect(row.health).toBe("active"); + // Health is current; the failure is history, and history is not cleared by + // the next good pass. + expect(row.lastError).toBeUndefined(); + expect(row.lastSyncFailureCode).toBe("GOOGLE_UNAVAILABLE"); + }); + + test("a skipped pass releases the claim and does not make an unsynced connection look synced", async () => { + const { t, connectionId } = await claimed(); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "skipped", + errorCode: "GOOGLE_RECONNECT_REQUIRED", + }); + const row = await readConnection(t, connectionId); + expect(row.syncStartedAt).toBeUndefined(); + expect(row.lastSyncAt).toBeUndefined(); + expect(row.gmail?.lastSyncedAt).toBeUndefined(); + expect(row.nextSyncAt).toBeGreaterThan(Date.now()); + }); + + test("an expired cursor is re-baselined and the gap is written down, not smoothed over", async () => { + const { t, connectionId } = await claimed(); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "synced", + historyId: "9999", + gapDetected: true, + }); + const row = await readConnection(t, connectionId); + expect(row.gmail?.historyId).toBe("9999"); + expect(row.health).toBe("active"); + expect(row.lastSyncFailureCode).toBe("GOOGLE_SYNC_GAP"); + expect(row.lastSyncFailure).toContain("not captured"); + }); + + test("a pass reporting after a disconnect changes nothing at all", async () => { + const { t, connectionId } = await claimed(); + await patchConnection(t, connectionId, { disconnectedAt: Date.now() }); + const accepted = await t.mutation( + internal.functions.googleSync.recordGoogleForwardSyncPass, + { connectionId, status: "synced", historyId: "4000" }, + ); + expect(accepted).toEqual({ accepted: false }); + expect((await readConnection(t, connectionId)).gmail?.historyId).toBeUndefined(); + }); +}); + +describe("a writer with nobody present still leaves a record", () => { + beforeEach(() => enableMailSync()); + + test("a pass that wrote days is audited against the person whose grant it used", async () => { + const { t, owner, connectionId, workspaceId } = await scenario(); + await patchConnection(t, connectionId, { syncStartedAt: Date.now() }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "synced", + historyId: "1200", + daysTouched: 2, + bytesWritten: 4096, + }); + const events = await t.run((ctx) => + ctx.db + .query("auditEvents") + .filter((q) => q.eq(q.field("action"), "google_sync_wrote")) + .collect(), + ); + expect(events).toHaveLength(1); + expect(events[0]!.workspaceId).toBe(workspaceId); + // Non-negotiable #4: the acting identity, not just the scope. Nobody is + // present, so it is the person whose grant every write was made on. + expect(events[0]!.actorUserId).toBe(owner); + expect(events[0]!.details).toMatchObject({ product: "gmail", days: 2, bytes: 4096 }); + // Counts only — never an address, a subject, or a path. + const written = JSON.stringify(events[0]); + expect(written).not.toContain("person@example.invalid"); + expect(written).not.toContain("0-inbox"); + }); + + test("a poll that found nothing is not an event", async () => { + const { t, connectionId } = await scenario(); + await patchConnection(t, connectionId, { syncStartedAt: Date.now() }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "synced", + historyId: "1200", + daysTouched: 0, + }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "skipped", + errorCode: "GOOGLE_RECONNECT_REQUIRED", + }); + /* + 288 rows a day per connection saying "looked, nothing there" is not an + audit trail, it is a way to lose the rows that matter inside one. + */ + const events = await t.run((ctx) => + ctx.db + .query("auditEvents") + .filter((q) => q.eq(q.field("action"), "google_sync_wrote")) + .collect(), + ); + expect(events).toHaveLength(0); + }); +}); + +describe("the pass re-asks every gate before it opens a credential", () => { + beforeEach(() => enableMailSync()); + + test("a disconnected account is skipped", async () => { + const { t, workspaceId, connectionId } = await scenario(); + await patchConnection(t, connectionId, { disconnectedAt: Date.now() }); + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { + workspaceId, + connectionId, + }), + ).toEqual({ kind: "skip", reason: "GOOGLE_DISCONNECTED" }); + }); + + test("a deployment that may not read mail is skipped even mid-flight", async () => { + const { t, workspaceId, connectionId } = await scenario(); + vi.stubEnv("MAIL_CONNECT_ENABLED", ""); + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { + workspaceId, + connectionId, + }), + ).toEqual({ kind: "skip", reason: "MAIL_CONNECT_DISABLED" }); + }); + + test("a grant Google has already refused is skipped rather than retried at it", async () => { + const { t, workspaceId, connectionId } = await scenario(); + await patchConnection(t, connectionId, { health: "reconnect_required" }); + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { + workspaceId, + connectionId, + }), + ).toEqual({ kind: "skip", reason: "GOOGLE_RECONNECT_REQUIRED" }); + }); + + test("a product turned off since the sweep looked is skipped", async () => { + const { t, workspaceId, connectionId } = await scenario(); + await patchConnection(t, connectionId, { products: [] }); + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { + workspaceId, + connectionId, + }), + ).toEqual({ kind: "skip", reason: "NO_SYNCABLE_PRODUCT" }); + }); + + test("a shared context is skipped, because a mailbox never lands in one", async () => { + const t = setupTest(); + const owner = await createUser(t, "owner@example.invalid"); + const shared = await createWorkspace(t, owner, "atlas-team", { kind: "shared" }); + const connectionId = await seedGoogleConnection(t, { workspaceId: shared, boundBy: owner }); + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { + workspaceId: shared, + connectionId, + }), + ).toEqual({ kind: "skip", reason: "NOT_PERSONAL_CONTEXT" }); + }); + + /* + ONE CONTEXT'S MAIL MUST NEVER BE WRITTEN INTO ANOTHER'S BUCKET. + + The pass takes a `workspaceId` (whose bucket credential is opened) and a + `connectionId` (whose mail is read). Nothing builds a mismatched pair + today — the sweep reads both off one row — but the pair is what a tenant + boundary is made of, and this is the only function that sees both. + Attacker and victim are in one database: two real personal contexts, each + with a real connection, so a refusal cannot come from the row not existing. + */ + test("a job whose workspace does not own the connection is refused outright", async () => { + const { t, workspaceId: victimWorkspace, connectionId: victimConnection } = await scenario(); + const attacker = await createUser(t, "attacker@example.invalid"); + const attackerWorkspace = await createWorkspace(t, attacker, "attacker"); + const attackerConnection = await seedGoogleConnection(t, { + workspaceId: attackerWorkspace, + boundBy: attacker, + address: "attacker@example.invalid", + }); + + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { + workspaceId: attackerWorkspace, + connectionId: victimConnection, + }), + ).toBeNull(); + // And the mirror image, so the check cannot be one-directional. + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { + workspaceId: victimWorkspace, + connectionId: attackerConnection, + }), + ).toBeNull(); + // Each still works against its own workspace, so the refusal above is the + // pairing and not something broken about either row. + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { + workspaceId: victimWorkspace, + connectionId: victimConnection, + }), + ).toMatchObject({ kind: "run" }); + }); + + test("...and a pass driven with a mismatched pair writes nothing and touches nothing", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1000" }); + const attacker = await createUser(t, "attacker@example.invalid"); + const attackerWorkspace = await createWorkspace(t, attacker, "attacker"); + const google = googleAndBucket({ backend }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await t.action(internal.functions.files.runFileOperation, { + workspaceId: attackerWorkspace, + scope: "private" as const, + operation: { kind: "googleForwardSync" as const, connectionId }, + }); + + expect(result).toMatchObject({ status: "skipped" }); + expect(google.calls).toEqual([]); + expect(backend.requests).toEqual([]); + const row = await readConnection(t, connectionId); + expect(row.gmail?.historyId).toBe("1000"); + expect(row.gmail?.lastSyncedAt).toBeUndefined(); + // The victim's own claim is untouched: another context's pass may not even + // release it. + expect(row.syncStartedAt).toBeTypeOf("number"); + expect(workspaceId).not.toEqual(attackerWorkspace); + }); + + test("a live connection is handed its settings and cursor, and no secret", async () => { + const { t, workspaceId, connectionId } = await scenario(); + await patchConnection(t, connectionId, { + gmail: { ...(await readConnection(t, connectionId)).gmail!, historyId: "1000" }, + }); + const job = await t.query(internal.functions.googleSync.googleForwardSyncJob, { + workspaceId, + connectionId, + }); + expect(job).toMatchObject({ + kind: "run", + product: "gmail", + address: "person@example.invalid", + mailboxSlug: "person-at-example-invalid", + destinationFolder: "0-inbox/email/person-at-example-invalid", + historyId: "1000", + }); + expect(JSON.stringify(job)).not.toContain("refresh"); + expect(JSON.stringify(job)).not.toContain("example-google-refresh-token-not-real"); + }); +}); + +describe("a pass that ran out of history pages", () => { + beforeEach(() => enableMailSync()); + + test("stays due immediately rather than waiting out its interval", async () => { + const { t, connectionId } = await scenario(); + await patchConnection(t, connectionId, { syncIntervalMinutes: 60, syncStartedAt: Date.now() }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "synced", + historyId: "1200", + catchUp: true, + }); + const row = await readConnection(t, connectionId); + // The cursor moved to the record boundary the walk actually reached, and + // the row says there is more behind it. + expect(row.gmail?.historyId).toBe("1200"); + expect(row.syncCatchUp).toBe(true); + expect(row.nextSyncAt).toBeLessThanOrEqual(Date.now()); + // An hourly connection is due on the very next tick, because the interval + // is about how often to *check* and this pass already knows there is work. + expect((await sweep(t)).started).toBe(1); + }); + + test("...and a pass that finished clears the flag, so it is not due forever", async () => { + const { t, connectionId } = await scenario(); + await patchConnection(t, connectionId, { + syncIntervalMinutes: 60, + syncCatchUp: true, + syncStartedAt: Date.now(), + }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "synced", + historyId: "1300", + }); + const row = await readConnection(t, connectionId); + expect(row.syncCatchUp).toBeUndefined(); + expect(row.nextSyncAt).toBe(row.lastSyncAt! + 60 * MINUTE); + expect((await sweep(t)).started).toBe(0); + }); + + test("the console says it is catching up rather than claiming it is current", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + await patchConnection(t, connectionId, { syncStartedAt: Date.now() }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "synced", + historyId: "1200", + catchUp: true, + }); + const view = ( + await asUser(t, owner).query(api.functions.googleConnect.listGoogleConnections, { + workspaceId, + }) + )[0]!; + expect(view.sync.catchingUp).toBe(true); + }); +}); + +describe("a grant Google has refused stays refused", () => { + beforeEach(() => enableMailSync()); + + test("a failed pass does not overwrite reconnect_required with a plain error", async () => { + const { t, connectionId } = await scenario(); + // The real sequence: minting marks the row, the pass then reports. + await t.mutation(internal.functions.googleConnect.markReconnectRequired, { connectionId }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "failed", + errorCode: "GOOGLE_RECONNECT_REQUIRED", + error: "Google needs to be reconnected before this mailbox can sync.", + }); + const row = await readConnection(t, connectionId); + expect(row.health).toBe("reconnect_required"); + // ...which is what makes the pass's own skip gate fire on the next tick, + // instead of asking Google for a token it has already refused, forever. + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { + workspaceId: row.workspaceId, + connectionId, + }), + ).toEqual({ kind: "skip", reason: "GOOGLE_RECONNECT_REQUIRED" }); + }); + + test("an ordinary failure still moves a healthy connection to error", async () => { + const { t, connectionId } = await scenario(); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "failed", + errorCode: "GOOGLE_UNAVAILABLE", + error: "Google did not answer reliably.", + }); + expect((await readConnection(t, connectionId)).health).toBe("error"); + }); + + test("repeated failures back off further each time, and a success resets the ladder", async () => { + const { t, connectionId } = await scenario(); + await patchConnection(t, connectionId, { syncIntervalMinutes: 5 }); + const waits: number[] = []; + for (let attempt = 0; attempt < 4; attempt += 1) { + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "failed", + errorCode: "GOOGLE_RATE_LIMITED", + error: "Google rate-limited this mailbox.", + }); + const row = await readConnection(t, connectionId); + waits.push(row.nextSyncAt! - row.lastSyncAt!); + } + // Strictly increasing: a connection Google is refusing is not asked again + // every fifteen minutes forever. + expect(waits[1]).toBeGreaterThan(waits[0]!); + expect(waits[2]).toBeGreaterThan(waits[1]!); + expect(waits[3]).toBeGreaterThan(waits[2]!); + expect((await readConnection(t, connectionId)).syncFailures).toBe(4); + + await patchConnection(t, connectionId, { syncStartedAt: Date.now() }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "synced", + historyId: "2200", + }); + const healthy = await readConnection(t, connectionId); + expect(healthy.syncFailures).toBeUndefined(); + expect(healthy.nextSyncAt).toBe(healthy.lastSyncAt! + 5 * MINUTE); + }); + + test("the ladder is capped, so a dead connection is still checked daily", async () => { + const { t, connectionId } = await scenario(); + await patchConnection(t, connectionId, { syncFailures: 40 }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "failed", + errorCode: "GOOGLE_RATE_LIMITED", + error: "Google rate-limited this mailbox.", + }); + const row = await readConnection(t, connectionId); + expect(row.nextSyncAt! - row.lastSyncAt!).toBeLessThanOrEqual(MAX_SYNC_BACKOFF_MS + MINUTE); + expect(row.nextSyncAt! - row.lastSyncAt!).toBeGreaterThanOrEqual(MAX_SYNC_BACKOFF_MS); + }); +}); + +describe("what the console is told, so the two states stop looking alike", () => { + beforeEach(() => enableMailSync()); + + async function listed(t: TestConvex, owner: Id<"users">, workspaceId: Id<"workspaces">) { + const rows = await asUser(t, owner).query(api.functions.googleConnect.listGoogleConnections, { + workspaceId, + }); + return rows[0]!; + } + + test("a connection that has never synced says so, and is due now", async () => { + const { t, owner, workspaceId } = await scenario(); + const view = await listed(t, owner, workspaceId); + expect(view.sync).toMatchObject({ + everSynced: false, + intervalMinutes: DEFAULT_SYNC_INTERVAL_MINUTES, + }); + expect(view.sync.lastAttemptAt).toBeUndefined(); + expect(view.sync.nextDueAt).toBeUndefined(); + }); + + test("...and one that has synced reports when, and when it is next due", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + await patchConnection(t, connectionId, { syncIntervalMinutes: 30, syncStartedAt: Date.now() }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "synced", + historyId: "2000", + }); + const view = await listed(t, owner, workspaceId); + expect(view.sync.everSynced).toBe(true); + expect(view.sync.intervalMinutes).toBe(30); + expect(view.sync.nextDueAt).toBe(view.sync.lastAttemptAt! + 30 * MINUTE); + expect(view.gmail?.lastSyncedAt).toBeTypeOf("number"); + }); + + test("a pass that only ever failed is not reported as having synced", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "failed", + errorCode: "GOOGLE_ACCESS_REFUSED", + error: "Google refused access to this mailbox.", + }); + const view = await listed(t, owner, workspaceId); + expect(view.sync.everSynced).toBe(false); + expect(view.sync.lastAttemptAt).toBeTypeOf("number"); + expect(view.sync.lastFailureCode).toBe("GOOGLE_ACCESS_REFUSED"); + expect(view.syncStatus).toBe("error"); + }); + + test("the last failure is still visible after a later pass succeeded", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "failed", + errorCode: "GOOGLE_UNAVAILABLE", + error: "Google did not answer reliably.", + }); + await t.mutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "synced", + historyId: "2100", + }); + const view = await listed(t, owner, workspaceId); + expect(view.syncStatus).toBe("active"); + expect(view.sync.everSynced).toBe(true); + expect(view.sync.lastFailureCode).toBe("GOOGLE_UNAVAILABLE"); + }); + + test("a disconnected account has no next due time to show", async () => { + const { t, owner, workspaceId, connectionId } = await scenario(); + await patchConnection(t, connectionId, { + lastSyncAt: Date.now(), + nextSyncAt: Date.now() + 15 * MINUTE, + disconnectedAt: Date.now(), + }); + const view = await listed(t, owner, workspaceId); + expect(view.syncStatus).toBe("disconnected"); + expect(view.sync.nextDueAt).toBeUndefined(); + }); + + test("nobody but the owner sees any of it", async () => { + const { t, owner, workspaceId } = await scenario(); + const editor = await createUser(t, "editor@example.invalid"); + await addMember(t, workspaceId, editor, "editor", owner); + expect( + await asUser(t, editor).query(api.functions.googleConnect.listGoogleConnections, { + workspaceId, + }), + ).toEqual([]); + }); +}); + +/* -------------------------------------------------------------------------- */ +/* end to end, through the credential barrier */ +/* -------------------------------------------------------------------------- */ + +/** + * A Gmail that answers the three calls a forward pass makes, and an S3 that + * holds what it writes. + * + * Routed by host, so the pass exercises the real `S3Store` (real SigV4, real + * XML) and the real `gmailSync.js` at the same time. Nothing here reaches the + * network: `edge-runtime` has no DNS and a request to anything unrouted throws. + */ +function googleAndBucket(options: { + backend: MemoryS3; + profileHistoryId?: string; + history?: + | { messageIds: string[]; historyId: string } + | { expired: true } + /** + * A paged history, one entry per page, each carrying a record id — the + * shape that exposes whether the walk's page limit is handled. Without + * this the fixture could never return a `nextPageToken`, and paging was + * the part of the real client nothing exercised. + */ + | { pages: { recordId: string; messageIds?: string[] }[]; historyId: string }; + messages?: { id: string; date: string; subject: string; text: string }[]; +}) { + const calls: string[] = []; + const messages = options.messages ?? []; + const json = (body: unknown, status = 200) => + new Response(JSON.stringify(body), { + status, + headers: { "content-type": "application/json" }, + }); + const base64Url = (text: string) => { + const bytes = new TextEncoder().encode(text); + let binary = ""; + for (const byte of bytes) binary += String.fromCharCode(byte); + return btoa(binary).replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, ""); + }; + + const fetchImpl = async (input: URL | RequestInfo, init: RequestInit = {}) => { + const url = new URL(typeof input === "string" ? input : String(input)); + if (url.hostname !== "gmail.googleapis.com") { + return await options.backend.fetchImpl(input, init); + } + calls.push(url.pathname); + + if (url.pathname === "/gmail/v1/users/me/profile") { + return json({ historyId: options.profileHistoryId ?? "5000" }); + } + if (url.pathname === "/gmail/v1/users/me/history") { + if (options.history && "expired" in options.history) return json({ error: { code: 404 } }, 404); + if (options.history && "pages" in options.history) { + const index = Number(url.searchParams.get("pageToken") ?? "0"); + const page = options.history.pages[index]; + if (!page) return json({ history: [], historyId: options.history.historyId }); + const body: Record = { + history: [ + { + id: page.recordId, + messagesAdded: (page.messageIds ?? []).map((id) => ({ message: { id } })), + }, + ], + // Every page carries the MAILBOX head, which is the whole trap. + historyId: options.history.historyId, + }; + if (index + 1 < options.history.pages.length) body.nextPageToken = String(index + 1); + return json(body); + } + const history = options.history ?? { messageIds: [], historyId: "1000" }; + return json({ + history: history.messageIds.map((id) => ({ messagesAdded: [{ message: { id } }] })), + historyId: history.historyId, + }); + } + const single = /^\/gmail\/v1\/users\/me\/messages\/([^/]+)$/.exec(url.pathname); + if (single) { + const message = messages.find((candidate) => candidate.id === single[1]); + if (!message) return json({ error: { code: 404 } }, 404); + return json({ + id: message.id, + threadId: `thread-${message.id}`, + internalDate: String(Date.parse(message.date)), + payload: { + mimeType: "multipart/mixed", + headers: [ + { name: "From", value: "Sender Example " }, + { name: "To", value: "person@example.invalid" }, + { name: "Subject", value: message.subject }, + ], + parts: [ + { mimeType: "text/plain", body: { size: message.text.length, data: base64Url(message.text) } }, + ], + }, + }); + } + if (url.pathname === "/gmail/v1/users/me/messages") { + const query = url.searchParams.get("q") ?? ""; + const after = /after:(\d+)/.exec(query); + const before = /before:(\d+)/.exec(query); + const from = after ? Number(after[1]) * 1000 : -Infinity; + const to = before ? Number(before[1]) * 1000 : Infinity; + const matching = messages.filter((message) => { + const at = Date.parse(message.date); + return at >= from && at < to; + }); + return json({ + messages: matching.map((message) => ({ id: message.id, threadId: `thread-${message.id}` })), + resultSizeEstimate: matching.length, + }); + } + return json({ error: { code: 404 } }, 404); + }; + + return { fetchImpl, calls }; +} + +async function endToEnd(options: { + historyId?: string; + quotaBytes?: number; + storage?: "connected" | "missing"; +} = {}) { + const t = setupTest(); + const owner = await createUser(t, "owner@example.invalid"); + const workspaceId = await createWorkspace(t, owner, "atlas"); + const connectionId = await seedGoogleConnection(t, { workspaceId, boundBy: owner }); + const keyset = requireKeyset(); + + await t.run(async (ctx) => { + const row = (await ctx.db.get(connectionId))!; + await ctx.db.patch(connectionId, { + gmail: { + ...row.gmail!, + historyId: options.historyId, + quotaBytes: options.quotaBytes ?? 1_000_000_000, + }, + // A cached access token, so the pass never calls Google's token + // endpoint: what is under test here is the sync, not the refresh. + encryptedAccessToken: await encryptSecret("example-access-token-not-real", keyset, { + workspaceId, + }), + accessTokenExpiresAt: Date.now() + 30 * MINUTE, + syncStartedAt: Date.now(), + }); + if (options.storage !== "missing") { + await ctx.db.insert("storageBindings", { + workspaceId, + provider: FAKE_STORAGE.provider, + endpoint: FAKE_STORAGE.endpoint, + region: FAKE_STORAGE.region, + bucket: FAKE_STORAGE.bucket, + accessKeyId: FAKE_STORAGE.accessKeyId, + encryptedSecretAccessKey: await encryptSecret(FAKE_STORAGE.secretAccessKey, keyset, { + workspaceId, + }), + capabilities: { conditionalWrite: true }, + status: "connected" as const, + lastVerifiedAt: Date.now(), + boundBy: owner, + createdAt: Date.now(), + updatedAt: Date.now(), + }); + } + }); + + const backend = memoryS3(FAKE_STORAGE.bucket); + return { t, owner, workspaceId, connectionId, backend }; +} + +async function runPass(t: TestConvex, workspaceId: Id<"workspaces">, connectionId: Id<"googleConnections">) { + return await t.action(internal.functions.files.runFileOperation, { + workspaceId, + scope: "private" as const, + operation: { kind: "googleForwardSync" as const, connectionId }, + }); +} + +describe("one pass, end to end, through the credential barrier", () => { + beforeEach(() => enableMailSync()); + + test("a baselined connection is not reported as having synced mail", async () => { + const { t, owner, workspaceId, connectionId, backend } = await endToEnd(); + const google = googleAndBucket({ backend, profileHistoryId: "7000" }); + vi.stubGlobal("fetch", google.fetchImpl); + + await runPass(t, workspaceId, connectionId); + + const view = ( + await asUser(t, owner).query(api.functions.googleConnect.listGoogleConnections, { + workspaceId, + }) + )[0]!; + /* + A baseline pass reads no mail — that is what forward-only means — so it + must not turn "connected, nothing read yet" into a card that looks + identical to a mailbox syncing fine. It is the same defect the schedule + block was written to close, arriving one state later. + */ + expect(view.sync.everSynced).toBe(false); + expect(view.sync.cursorReady).toBe(true); + expect(view.gmail?.lastSyncedAt).toBeUndefined(); + + // ...and an ordinary pass afterwards, even one that finds nothing new, is + // a real sync: the mailbox was actually read. + await patchConnection(t, connectionId, { syncStartedAt: Date.now() }); + const quiet = googleAndBucket({ backend, history: { messageIds: [], historyId: "7100" } }); + vi.stubGlobal("fetch", quiet.fetchImpl); + await runPass(t, workspaceId, connectionId); + const after = ( + await asUser(t, owner).query(api.functions.googleConnect.listGoogleConnections, { + workspaceId, + }) + )[0]!; + expect(after.sync.everSynced).toBe(true); + }); + + test("re-baselining after a gap does not invent a sync that never read anything", async () => { + const { t, owner, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1000" }); + const google = googleAndBucket({ backend, history: { expired: true }, profileHistoryId: "8000" }); + vi.stubGlobal("fetch", google.fetchImpl); + + await runPass(t, workspaceId, connectionId); + + const view = ( + await asUser(t, owner).query(api.functions.googleConnect.listGoogleConnections, { + workspaceId, + }) + )[0]!; + expect(view.sync.everSynced).toBe(false); + expect(view.sync.lastFailureCode).toBe("GOOGLE_SYNC_GAP"); + }); + + test("a connection with no cursor takes a baseline and fetches no mail", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd(); + const google = googleAndBucket({ backend, profileHistoryId: "7000" }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ kind: "googleForwardSync", status: "synced", daysTouched: 0 }); + expect(google.calls).toEqual(["/gmail/v1/users/me/profile"]); + const row = await readConnection(t, connectionId); + expect(row.gmail?.historyId).toBe("7000"); + expect(row.syncStartedAt).toBeUndefined(); + // Forward-only: a baseline is where syncing starts, not a licence to read + // what came before it. + expect(Object.keys(backend.snapshot())).toEqual([]); + }); + + test("a cursor advances, and the day the mail landed on is written into the bucket", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1000" }); + const google = googleAndBucket({ + backend, + history: { messageIds: ["msg-1"], historyId: "1100" }, + messages: [ + { + id: "msg-1", + date: "2026-09-08T09:14:00.000Z", + subject: "Quarterly numbers", + text: "The numbers are attached.", + }, + ], + }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ + kind: "googleForwardSync", + status: "synced", + daysTouched: 1, + cursorAdvanced: true, + }); + const row = await readConnection(t, connectionId); + expect(row.gmail?.historyId).toBe("1100"); + expect(row.gmail?.lastSyncedAt).toBeTypeOf("number"); + expect(row.syncBytesWritten).toBeGreaterThan(0); + + const written = backend.snapshot(); + const day = "0-inbox/email/person-at-example-invalid/2026-09-08.md"; + expect(Object.keys(written)).toContain(day); + expect(written[day]).toContain("Quarterly numbers"); + }); + + test("re-running the same pass writes no new bytes and moves nothing", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1000" }); + const google = googleAndBucket({ + backend, + history: { messageIds: ["msg-1"], historyId: "1100" }, + messages: [ + { + id: "msg-1", + date: "2026-09-08T09:14:00.000Z", + subject: "Quarterly numbers", + text: "The numbers are attached.", + }, + ], + }); + vi.stubGlobal("fetch", google.fetchImpl); + + await runPass(t, workspaceId, connectionId); + const first = backend.snapshot(); + + // Wind the cursor back and run it again. The day is rebuilt from Gmail's + // live state either way, and `renderChannelDayNote`'s `updated` is keyed + // to the newest message rather than to wall-clock time — so a repeated + // pass writes the same bytes. That is the property that makes a scheduled + // loop safe to run every five minutes forever. + await t.run(async (ctx) => { + const row = (await ctx.db.get(connectionId))!; + await ctx.db.patch(connectionId, { + gmail: { ...row.gmail!, historyId: "1000" }, + syncStartedAt: Date.now(), + }); + }); + await runPass(t, workspaceId, connectionId); + expect(backend.snapshot()).toEqual(first); + }); + + test("a walk that runs out of pages stores the record boundary, not the head", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1000" }); + // Fifty-one pages against a fifty-page walk: the client stops one short of + // the end, and every page has claimed the mailbox head all along. + const pages = Array.from({ length: 51 }, (_, index) => ({ + recordId: String(1100 + index), + })); + const google = googleAndBucket({ backend, history: { pages, historyId: "999999" } }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ status: "synced", truncated: true }); + const row = await readConnection(t, connectionId); + /* + The head would say "caught up" while fifty-one pages of history had been + read and everything behind them skipped forever. The last record actually + walked says where to resume, and the row says there is more. + */ + expect(row.gmail?.historyId).toBe("1149"); + expect(row.gmail?.historyId).not.toBe("999999"); + expect(row.syncCatchUp).toBe(true); + expect(row.nextSyncAt).toBeLessThanOrEqual(Date.now()); + expect((await sweep(t)).started).toBe(1); + }); + + test("...and the next pass carries on from there rather than repeating itself", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1149" }); + const google = googleAndBucket({ + backend, + history: { pages: [{ recordId: "1150" }], historyId: "999999" }, + }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ status: "synced", truncated: false }); + const row = await readConnection(t, connectionId); + // A walk that reached the end may store the head, and only then. + expect(row.gmail?.historyId).toBe("999999"); + expect(row.syncCatchUp).toBeUndefined(); + }); + + test("an expired cursor re-baselines forward and records the gap", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1000" }); + const google = googleAndBucket({ + backend, + history: { expired: true }, + profileHistoryId: "8000", + }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ status: "synced", gapDetected: true }); + const row = await readConnection(t, connectionId); + expect(row.gmail?.historyId).toBe("8000"); + expect(row.lastSyncFailureCode).toBe("GOOGLE_SYNC_GAP"); + }); + + test("bytes written before a quota ceiling are counted, or the ceiling never arrives", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ + historyId: "1000", + // Room for the first day and not the second: the pass writes, then stops. + quotaBytes: 2_400, + }); + const google = googleAndBucket({ + backend, + history: { messageIds: ["msg-1", "msg-2"], historyId: "1100" }, + messages: [ + { + id: "msg-1", + date: "2026-09-08T09:14:00.000Z", + subject: "Quarterly numbers", + text: "The numbers are attached.", + }, + { + id: "msg-2", + date: "2026-09-09T09:14:00.000Z", + subject: "Follow-up", + text: "And the follow-up.", + }, + ], + }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ status: "failed", errorCode: "MAIL_QUOTA_EXCEEDED" }); + const row = await readConnection(t, connectionId); + /* + The whole point of the ceiling. A failed pass that wrote real bytes and + then dropped the count leaves `bytesAlreadyUsed` frozen below the limit, + so every later pass re-writes the same day, re-counts nothing, and the + ceiling is never crossed — a quota that cannot be reached is decoration. + */ + expect(row.syncBytesWritten).toBeGreaterThan(0); + expect(row.gmail?.historyId).toBe("1000"); + }); + + test("a quota ceiling stops the pass and does NOT advance the cursor past unwritten mail", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ + historyId: "1000", + quotaBytes: 1, + }); + const google = googleAndBucket({ + backend, + history: { messageIds: ["msg-1"], historyId: "1100" }, + messages: [ + { + id: "msg-1", + date: "2026-09-08T09:14:00.000Z", + subject: "Quarterly numbers", + text: "The numbers are attached.", + }, + ], + }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ status: "failed", errorCode: "MAIL_QUOTA_EXCEEDED" }); + const row = await readConnection(t, connectionId); + expect(row.gmail?.historyId).toBe("1000"); + expect(row.errorCode).toBe("MAIL_QUOTA_EXCEEDED"); + }); + + test.each([ + [429, {}, "GOOGLE_RATE_LIMITED"], + [403, { reason: "userRateLimitExceeded" }, "GOOGLE_RATE_LIMITED"], + [403, {}, "GOOGLE_ACCESS_REFUSED"], + [503, {}, "GOOGLE_UNAVAILABLE"], + [418, {}, "GMAIL_HTTP_418"], + ])( + "Gmail answering %i is classified, not flattened into one sentence", + async (status, detail, expected) => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1000" }); + const google = googleAndBucket({ backend }); + vi.stubGlobal("fetch", async (input: URL | RequestInfo, init: RequestInit = {}) => { + const url = new URL(typeof input === "string" ? input : String(input)); + if (url.hostname === "gmail.googleapis.com") { + return new Response( + JSON.stringify({ + error: { + code: status, + message: "denied", + errors: "reason" in detail ? [{ reason: detail.reason }] : undefined, + }, + }), + { status, headers: { "content-type": "application/json" } }, + ); + } + return await google.fetchImpl(input, init); + }); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ status: "failed", errorCode: expected }); + const row = await readConnection(t, connectionId); + expect(row.lastSyncFailureCode).toBe(expected); + // The advice has to match the cause: telling somebody to reconnect and + // re-approve Gmail for a rate limit that clears itself is worse than + // saying nothing. + if (expected === "GOOGLE_RATE_LIMITED" || expected === "GOOGLE_UNAVAILABLE") { + expect(row.lastSyncFailure).not.toContain("Reconnect"); + } + // Nothing that failed advanced the cursor. + expect(row.gmail?.historyId).toBe("1000"); + }, + ); + + test("Gmail refusing the account is recorded as a failure a person can read", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1000" }); + const google = googleAndBucket({ backend }); + vi.stubGlobal("fetch", async (input: URL | RequestInfo, init: RequestInit = {}) => { + const url = new URL(typeof input === "string" ? input : String(input)); + if (url.hostname === "gmail.googleapis.com") { + return new Response(JSON.stringify({ error: { code: 403, message: "denied" } }), { + status: 403, + headers: { "content-type": "application/json" }, + }); + } + return await google.fetchImpl(input, init); + }); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ status: "failed", errorCode: "GOOGLE_ACCESS_REFUSED" }); + const row = await readConnection(t, connectionId); + expect(row.gmail?.historyId).toBe("1000"); + expect(row.health).toBe("error"); + expect(row.lastSyncFailureCode).toBe("GOOGLE_ACCESS_REFUSED"); + }); + + test("a context with no bucket records the failure rather than throwing into the scheduler", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ + historyId: "1000", + storage: "missing", + }); + const google = googleAndBucket({ backend }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ status: "failed", errorCode: "STORAGE_NOT_CONNECTED" }); + const row = await readConnection(t, connectionId); + expect(row.syncStartedAt).toBeUndefined(); + // Nothing was asked of Google, because nothing could have been written. + expect(google.calls).toEqual([]); + }); + + test("a pass that cannot mint a token records it, rather than stranding the claim", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1000" }); + // No cached token to fall back on and nothing configured to mint a new + // one. What matters is not which of the two codes comes back but that the + // pass *reports*: a throw escaping here would leave the row claimed and + // silent for fifteen minutes with nothing on it to explain why. + await t.run((ctx) => + ctx.db.patch(connectionId, { + encryptedAccessToken: undefined, + accessTokenExpiresAt: undefined, + }), + ); + vi.stubEnv("GOOGLE_OAUTH_CLIENT_ID", ""); + const google = googleAndBucket({ backend }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ status: "failed", errorCode: "GOOGLE_RECONNECT_REQUIRED" }); + const row = await readConnection(t, connectionId); + expect(row.syncStartedAt).toBeUndefined(); + expect(row.gmail?.historyId).toBe("1000"); + expect(row.lastSyncFailureCode).toBe("GOOGLE_RECONNECT_REQUIRED"); + // The sentence shown never quotes this deployment's own configuration. + expect(row.lastSyncFailure).not.toContain("client"); + expect(google.calls).toEqual([]); + }); + + test("a disconnected account releases its claim without touching Google or the bucket", async () => { + const { t, workspaceId, connectionId, backend } = await endToEnd({ historyId: "1000" }); + await patchConnection(t, connectionId, { disconnectedAt: Date.now() }); + const google = googleAndBucket({ backend }); + vi.stubGlobal("fetch", google.fetchImpl); + + const result = await runPass(t, workspaceId, connectionId); + + expect(result).toMatchObject({ status: "skipped", errorCode: "GOOGLE_DISCONNECTED" }); + expect(google.calls).toEqual([]); + expect(backend.requests).toEqual([]); + }); +}); diff --git a/apps/convex/_generated/api.d.ts b/apps/convex/_generated/api.d.ts index 150e0c9e0..789cedf45 100644 --- a/apps/convex/_generated/api.d.ts +++ b/apps/convex/_generated/api.d.ts @@ -59,6 +59,7 @@ import type * as functions_lib_nameClaims from "../functions/lib/nameClaims.js"; import type * as functions_lib_names from "../functions/lib/names.js"; import type * as functions_lib_noteCount from "../functions/lib/noteCount.js"; import type * as functions_lib_noteLinks from "../functions/lib/noteLinks.js"; +import type * as functions_lib_googleSchedule from "../functions/lib/googleSchedule.js"; import type * as functions_lib_privacy from "../functions/lib/privacy.js"; import type * as functions_lib_rateLimit from "../functions/lib/rateLimit.js"; import type * as functions_lib_scaffold from "../functions/lib/scaffold.js"; @@ -68,6 +69,7 @@ import type * as functions_lib_verification from "../functions/lib/verification. import type * as functions_lib_workspaceAuth from "../functions/lib/workspaceAuth.js"; import type * as functions_calendarConnect from "../functions/calendarConnect.js"; import type * as functions_googleConnect from "../functions/googleConnect.js"; +import type * as functions_googleSync from "../functions/googleSync.js"; import type * as functions_meetings_transcribe from "../functions/meetings/transcribe.js"; import type * as functions_names from "../functions/names.js"; import type * as functions_provisioning from "../functions/provisioning.js"; @@ -136,6 +138,7 @@ declare const fullApi: ApiFromModules<{ "functions/lib/names": typeof functions_lib_names; "functions/lib/noteCount": typeof functions_lib_noteCount; "functions/lib/noteLinks": typeof functions_lib_noteLinks; + "functions/lib/googleSchedule": typeof functions_lib_googleSchedule; "functions/lib/privacy": typeof functions_lib_privacy; "functions/lib/rateLimit": typeof functions_lib_rateLimit; "functions/lib/scaffold": typeof functions_lib_scaffold; @@ -145,6 +148,7 @@ declare const fullApi: ApiFromModules<{ "functions/lib/workspaceAuth": typeof functions_lib_workspaceAuth; "functions/calendarConnect": typeof functions_calendarConnect; "functions/googleConnect": typeof functions_googleConnect; + "functions/googleSync": typeof functions_googleSync; "functions/meetings/transcribe": typeof functions_meetings_transcribe; "functions/names": typeof functions_names; "functions/provisioning": typeof functions_provisioning; diff --git a/apps/convex/crons.ts b/apps/convex/crons.ts index 22c9e2840..7afeb2d4f 100644 --- a/apps/convex/crons.ts +++ b/apps/convex/crons.ts @@ -1,10 +1,25 @@ /** - * Scheduled maintenance. + * Scheduled maintenance, and — since 2026-09-10 — one scheduled engine. * * Nothing here may hold a decision. A cron is the wrong place for anything a - * person would want to see refused in the moment, so this file is limited to - * jobs whose only effect is that the database stops accumulating things nobody - * reads. + * person would want to see refused in the moment, and that rule is unchanged. + * + * What has changed is the second half of the original sentence. This file was + * written for jobs whose only effect is that the database stops accumulating + * things nobody reads, and it later gained a job that *restarts* work over a + * disposable derivative (`restart stalled search backfills`) — still a repair, + * still nothing a customer's bucket would notice. `sync due Google accounts` + * is a third kind and it should be named rather than filed quietly beside the + * sweeps: it calls a third party on a customer's quota and writes canonical + * Markdown into their bucket, on a clock, with nobody present. + * + * That is admitted here rather than argued away, because the honest limit on + * this file is not "only deletions" — it is that **every job here must hold no + * decision, and a job that acts outside this database must additionally + * re-ask, at the moment it acts, everything that could have changed since it + * was scheduled**. The sync job's own comment below makes that case in full, + * and `googleForwardSyncJob` is where the re-asking lives. A fourth job of + * this kind owes the same two paragraphs. */ import { cronJobs } from "convex/server"; @@ -146,4 +161,46 @@ crons.interval( {}, ); +/** + * Poll every connected Google account that is due. + * + * **The second job here that starts work rather than deleting it**, and it owes + * the same argument the search sweep above makes. It holds no decision: + * whether a connection may sync at all — is it still connected, is its context + * personal, does it enable a product this engine can advance, does this + * deployment allow reading a restricted scope — is the connection's own state, + * re-asked by `googleForwardSyncJob` inside the pass, before a credential is + * opened. What this decides is only *when to look*. + * + * One thing is genuinely this job's alone and is not re-asked inside the pass: + * **due-ness**. The sweep decides a connection is due and the pass then syncs + * without asking whether it should have waited. That is deliberate — the pass + * runs minutes later and re-deciding due-ness against a clock that has moved + * would make a claimed pass refuse itself — but it is worth stating plainly + * rather than letting "every gate is re-asked" imply more than is true. + * + * It exists because nothing else ever looked. Connecting a mailbox recorded a + * grant and a `historyId` and then nothing advanced it: no cron, no webhook — + * Google's push path is deliberately not built (`docs/decisions/communications.md`) + * — and the gateway's `scheduled()` handler has no trigger configured. A + * person connected Gmail and their mail never arrived. + * + * **Five minutes because that is the floor**, not because every account is + * polled that often. `syncIntervalMinutes` is per connection and defaults to + * fifteen; the sweep starts a pass only where `now >= lastSyncAt + interval`, + * which is how one fixed tick serves many different frequencies. Below five + * the tick would be finer than the shortest interval anybody may choose, and + * every extra tick is a transaction that reads an index to find nothing. + * + * And, as above: each run only touches connections nothing has written to in + * fifteen minutes, so a pass that is still running is never overtaken by a + * second one. + */ +crons.interval( + "sync due Google accounts", + { minutes: 5 }, + internal.functions.googleSync.sweepDueGoogleSyncs, + {}, +); + export default crons; diff --git a/apps/convex/functions/files.ts b/apps/convex/functions/files.ts index 45e89f799..ce9311a40 100644 --- a/apps/convex/functions/files.ts +++ b/apps/convex/functions/files.ts @@ -85,6 +85,22 @@ import { storeForBinding } from "../../mcp/src/store/factory.js"; // which is Convex's runtime too. It holds the write token for the life of one // call and puts it in exactly one place, an `Authorization` header. import { createD1Client } from "../../mcp/src/search/d1/client.js"; +/* + * The Gmail pipeline, imported rather than ported, for exactly the reason the + * two imports above are: `apps/mcp` targets the Workers runtime, which is + * Convex's runtime too, and this module takes its socket, its access token and + * its store as parameters — it opens nothing itself. + * + * It came back with the forward sync loop. #388 removed the historical + * backfill that used to import it and left the module reachable from nothing + * at all, which is how a complete, fixture-tested mail pipeline sat in the + * repository while connected mailboxes synced nothing. + */ +import { + getProfileHistoryId, + GmailApiError, + runIncrementalSync, +} from "../../mcp/src/communications/gmailSync.js"; import { D1_ACCOUNT_SECRET, D1_TOKEN_SECRET, @@ -465,6 +481,25 @@ const googleSyncRunValidator = v.object({ continue: v.boolean(), }); +/** + * One forward sync pass, as the scheduler sees it. No mail, no path, no + * cursor — the cursor is written to the connection row by + * `recordGoogleForwardSyncPass`, and a scheduled action's return value is read + * by nobody but a test. + */ +const googleForwardSyncValidator = v.object({ + kind: v.literal("googleForwardSync"), + connectionId: v.id("googleConnections"), + status: v.union(v.literal("synced"), v.literal("skipped"), v.literal("failed")), + daysTouched: v.number(), + bytesWritten: v.number(), + cursorAdvanced: v.boolean(), + gapDetected: v.boolean(), + /** The history walk ran out of pages; this connection has more to drain. */ + truncated: v.boolean(), + errorCode: v.optional(v.string()), +}); + const operationResultValidator = v.union( listingValidator, fileValidator, @@ -481,6 +516,7 @@ const operationResultValidator = v.union( indexMaintainedValidator, indexProjectedValidator, googleSyncRunValidator, + googleForwardSyncValidator, ); const operationValidator = v.union( @@ -531,6 +567,13 @@ const operationValidator = v.union( */ v.object({ kind: v.literal("projectIndex"), passes: v.optional(v.number()) }), v.object({ kind: v.literal("googleGmailBackfill"), runId: v.id("googleSyncRuns") }), + /** + * Advance one connected Google account from its own cursor. Scheduled by + * `googleSync.sweepDueGoogleSyncs` and by nothing else — there is no public + * action that reaches this variant, and no argument on it a caller could use + * to name a context: the workspace comes from the connection row. + */ + v.object({ kind: v.literal("googleForwardSync"), connectionId: v.id("googleConnections") }), v.object({ kind: v.literal("write"), path: v.string(), @@ -660,6 +703,17 @@ type OperationResult = bytesWritten: number; continue: boolean; } + | { + kind: "googleForwardSync"; + connectionId: Id<"googleConnections">; + status: "synced" | "skipped" | "failed"; + daysTouched: number; + bytesWritten: number; + cursorAdvanced: boolean; + gapDetected: boolean; + truncated: boolean; + errorCode?: string; + } | { kind: "listing"; path: string; @@ -924,12 +978,66 @@ export const runFileOperation = internalAction({ ); } + /* + * A FORWARD SYNC PASS ASKS THE ROW BEFORE IT ASKS FOR A CREDENTIAL. + * + * Same ordering, same reason, as the projection pass above. The sweep that + * scheduled this holds no decision and ran minutes ago; in between, the + * account can have been disconnected, its product turned off, or its grant + * refused by Google. Asking first means none of those decrypt a customer's + * storage secret on the way to doing nothing. + * + * A `null` job means there is no connection row to report against at all, + * so there is also no claim to release. + */ + let forwardSyncJob: ForwardSyncJob = null; + if (args.operation.kind === "googleForwardSync") { + forwardSyncJob = await ctx.runQuery( + internal.functions.googleSync.googleForwardSyncJob, + { workspaceId: args.workspaceId, connectionId: args.operation.connectionId }, + ); + if (forwardSyncJob === null) { + /* + * No row this workspace owns — it was deleted, or the pair of + * arguments does not agree (see `googleForwardSyncJob`). Nothing is + * written, and in particular the *other* context's row is not: a + * mismatched pair that released somebody else's claim and pushed their + * next sync out would be a cross-tenant write, small but real. + */ + return { + kind: "googleForwardSync", + connectionId: args.operation.connectionId, + status: "skipped", + daysTouched: 0, + bytesWritten: 0, + cursorAdvanced: false, + gapDetected: false, + truncated: false, + }; + } + if (forwardSyncJob.kind === "skip") { + return await releaseForwardSync( + ctx, + args.operation.connectionId, + forwardSyncJob.reason, + ); + } + } + let credential: GatewayCredential | null; try { credential = await ctx.runAction(internal.functions.storage.getBindingForGateway, { workspaceId: args.workspaceId, }); } catch { + if (args.operation.kind === "googleForwardSync") { + return await failForwardSync( + ctx, + args.operation.connectionId, + "STORAGE_UNUSABLE", + "This context's bucket configuration could not be used. Reconnect storage.", + ); + } throw new ConvexError({ code: "STORAGE_UNUSABLE", message: @@ -937,6 +1045,14 @@ export const runFileOperation = internalAction({ }); } if (credential === null) { + if (args.operation.kind === "googleForwardSync") { + return await failForwardSync( + ctx, + args.operation.connectionId, + "STORAGE_NOT_CONNECTED", + "This context has no bucket connected yet. Connect storage before syncing Google.", + ); + } throw new ConvexError({ code: "STORAGE_NOT_CONNECTED", message: @@ -978,6 +1094,14 @@ export const runFileOperation = internalAction({ // The constructor's message can quote the endpoint the customer typed. // Nothing it says helps here, and re-throwing it would put provider text // in front of the user with no way to know what else is in it. + if (args.operation.kind === "googleForwardSync") { + return await failForwardSync( + ctx, + args.operation.connectionId, + "STORAGE_UNUSABLE", + "This context's bucket configuration could not be used. Reconnect storage.", + ); + } throw new ConvexError({ code: "STORAGE_UNUSABLE", message: @@ -985,6 +1109,10 @@ export const runFileOperation = internalAction({ }); } + if (args.operation.kind === "googleForwardSync" && forwardSyncJob?.kind === "run") { + return await runGoogleForwardSync(ctx, store, forwardSyncJob); + } + const result = await executeOperation( store, args.scope, @@ -1093,6 +1221,363 @@ function timeoutFetch( return globalThis.fetch(input, timeout ? { ...init, signal: timeout } : init); } +/* -------------------------------------------------------------------------- */ +/* the forward sync pass, one connection */ +/* -------------------------------------------------------------------------- */ + +/** + * What `googleSync.googleForwardSyncJob` answered. + * + * Written out rather than inferred because the inference would run through + * `internal.functions.googleSync`, which is the cycle every annotated handler + * in this file exists to avoid. + */ +type ForwardSyncJob = + | null + | { kind: "skip"; reason: string } + | { + kind: "run"; + connectionId: Id<"googleConnections">; + product: "gmail"; + address: string; + mailboxSlug: string; + destinationFolder: string; + folders: ("inbox" | "sent")[]; + quotaBytes: number; + bytesAlreadyUsed: number; + attachmentMode: "metadata-only" | "store"; + attachmentRetentionDays?: number | "forever"; + historyId?: string; + }; + +type ForwardSyncResult = Extract; + +/** + * Nothing to do, and the claim released. + * + * A skipped pass must leave `lastSyncAt` alone — a connection that has never + * synced and one whose pass was skipped are the same connection, and making + * the second look synced is precisely the confusion this whole loop exists to + * remove. + */ +async function releaseForwardSync( + ctx: ActionCtx, + connectionId: Id<"googleConnections">, + reason: string | undefined, +): Promise { + await ctx.runMutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "skipped", + errorCode: reason, + }); + return { + kind: "googleForwardSync", + connectionId, + status: "skipped", + daysTouched: 0, + bytesWritten: 0, + cursorAdvanced: false, + gapDetected: false, + truncated: false, + errorCode: reason, + }; +} + +/** A pass that could not run, recorded where the owner can read it. */ +async function failForwardSync( + ctx: ActionCtx, + connectionId: Id<"googleConnections">, + errorCode: string, + error: string, +): Promise { + await ctx.runMutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId, + status: "failed", + errorCode, + error, + }); + return { + kind: "googleForwardSync", + connectionId, + status: "failed", + daysTouched: 0, + bytesWritten: 0, + cursorAdvanced: false, + gapDetected: false, + truncated: false, + errorCode, + }; +} + +const GMAIL_RATE_LIMIT_REASONS = new Set([ + "dailyLimitExceeded", + "rateLimitExceeded", + "userRateLimitExceeded", + "quotaExceeded", +]); + +/** + * Turn whatever went wrong into a code and a sentence a person can act on. + * + * Trimmed from the classifier #388 removed with the historical backfill: the + * retry ladder went with it (a forward pass is retried by the sweep on its own + * interval, with `SYNC_FAILURE_BACKOFF_MS` as the floor), but the + * classification did not, because "Google refused this account" and "Google + * was briefly unavailable" are still different sentences to show somebody. + */ +function classifyForwardSyncError(error: unknown): { code: string; message: string } { + if (error instanceof GmailApiError) { + const reason = typeof error.reason === "string" ? error.reason : undefined; + const googleStatus = typeof error.googleStatus === "string" ? error.googleStatus : undefined; + if ( + error.status === 429 || + (error.status === 403 && + (GMAIL_RATE_LIMIT_REASONS.has(reason ?? "") || googleStatus === "RESOURCE_EXHAUSTED")) + ) { + return { + code: "GOOGLE_RATE_LIMITED", + message: "Google rate-limited this mailbox. The next scheduled pass will try again.", + }; + } + if (error.status === 401 || error.status === 403) { + return { + code: "GOOGLE_ACCESS_REFUSED", + message: "Google refused access to this mailbox. Reconnect the account and approve Gmail access.", + }; + } + if (error.status >= 500) { + return { + code: "GOOGLE_UNAVAILABLE", + message: "Google did not answer reliably. The next scheduled pass will try again.", + }; + } + return { + code: `GMAIL_HTTP_${error.status}`, + message: `Gmail answered with ${error.status}. The next scheduled pass will try again.`, + }; + } + if ( + error instanceof Error && + (error.name === "AbortError" || error.name === "TimeoutError") + ) { + return { + code: "GOOGLE_SYNC_TIMEOUT", + message: "Gmail or storage took too long. The next scheduled pass resumes from the same cursor.", + }; + } + return { + code: "GOOGLE_SYNC_FAILED", + message: "This mailbox did not sync. The next scheduled pass resumes from the same cursor.", + }; +} + +/** + * ONE FORWARD PASS: advance this connection's cursor, write whatever changed. + * + * Forward-only, per #388 and `docs/decisions/communications.md`. Three shapes: + * + * - **No cursor yet.** The connection was bound before a baseline could be + * read, so one is taken now from `users.getProfile` and stored. Nothing is + * fetched: forward-only means the mail from before this moment is not this + * loop's to collect. + * - **A cursor.** `history.list` from it, rebuild every day a changed message + * landed on from Gmail's live state, store the new cursor. + * - **An expired cursor.** Gmail's 404 comes back as `gapDetected` rather + * than an error. The documented recovery was a reconcile over the backfill + * window, which forward-only does not have — so the cursor is re-baselined + * and the gap is recorded on the row as a failure a person can read. + * + * **The cursor is never advanced past mail that was not written.** A quota + * ceiling reached mid-pass, or anything thrown, leaves `historyId` exactly + * where it was, so the next pass asks Gmail the same question again. Advancing + * it would be the one bug in this file that loses somebody's mail silently. + */ +async function runGoogleForwardSync( + ctx: ActionCtx, + store: FileStore, + job: Extract, +): Promise { + try { + /* + * Inside the try, deliberately. Minting can *throw* as well as answer + * `null` — a deployment with no Google client id configured, an envelope + * that will not open — and a throw that escapes this function leaves the + * scheduler holding the failure and the row holding its claim, so the + * connection goes quiet for fifteen minutes with nothing on it to say why. + */ + const minted = await ctx.runAction(internal.functions.googleConnect.mintGoogleAccessToken, { + connectionId: job.connectionId, + }); + if (minted === null) { + // `mintGoogleAccessToken` has already marked the row + // `reconnect_required` if Google refused the grant outright; this + // records the pass itself. + return await failForwardSync( + ctx, + job.connectionId, + "GOOGLE_RECONNECT_REQUIRED", + "Google needs to be reconnected before this mailbox can sync.", + ); + } + + if (job.historyId === undefined) { + const historyId = await getProfileHistoryId({ + fetchImpl: timeoutFetch, + accessToken: minted.accessToken, + }); + await ctx.runMutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId: job.connectionId, + status: "synced", + historyId, + daysTouched: 0, + bytesWritten: 0, + // A cursor, not a sync: this pass read no mail, and the console must + // be able to say so rather than showing a mailbox that looks current. + baseline: true, + }); + return { + kind: "googleForwardSync", + connectionId: job.connectionId, + status: "synced", + daysTouched: 0, + bytesWritten: 0, + cursorAdvanced: true, + gapDetected: false, + truncated: false, + }; + } + + const result = await runIncrementalSync({ + store, + fetchImpl: timeoutFetch, + accessToken: minted.accessToken, + mailboxSlug: job.mailboxSlug, + address: job.address, + folders: job.folders, + startHistoryId: job.historyId, + folder: job.destinationFolder, + // The same nonce the backfill used, so a day rewritten by either path + // keeps its message anchors — see `packages/communications/src/note.js`. + nonce: `gmail:${job.connectionId}`, + /* + * NO `now`, AND THAT IS THE WHOLE POINT OF A LOOP THAT REPEATS. + * + * `renderDay` defaults `updated` to the latest message's own `sentAt` + * precisely so that re-rendering an unchanged day is byte-identical, and + * `syncOneDay` forwards whatever `now` a caller passes straight through + * to it. The backfill removed by #388 passed a wall clock, which was + * survivable for a one-shot import and is not for a pass that runs every + * few minutes: every touched day would get a new `updated`, a new write, + * and a new etag, forever — churn wearing the costume of sync activity. + * Attachment retention is measured against a real wall clock inside + * `syncOneDay` regardless, so nothing here loses a clock it needed. + */ + quotaBytes: job.quotaBytes, + bytesAlreadyUsed: job.bytesAlreadyUsed, + attachmentMode: job.attachmentMode, + attachmentRetentionDays: job.attachmentRetentionDays, + }); + + if (result.gapDetected) { + const historyId = await getProfileHistoryId({ + fetchImpl: timeoutFetch, + accessToken: minted.accessToken, + }); + await ctx.runMutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId: job.connectionId, + status: "synced", + historyId, + daysTouched: 0, + bytesWritten: 0, + gapDetected: true, + // Re-baselining reads nothing either, for the same reason. + baseline: true, + }); + return { + kind: "googleForwardSync", + connectionId: job.connectionId, + status: "synced", + daysTouched: 0, + bytesWritten: 0, + cursorAdvanced: true, + gapDetected: true, + truncated: false, + }; + } + + if (result.quotaExceeded) { + // Whatever was written stays written and is counted; the cursor does + // not move, so the days this pass could not afford are asked for again + // once the connection has room. + await ctx.runMutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId: job.connectionId, + status: "failed", + daysTouched: result.daysTouched.length, + bytesWritten: result.bytesWritten, + errorCode: "MAIL_QUOTA_EXCEEDED", + error: "This connection reached its storage quota before the pass finished.", + }); + return { + kind: "googleForwardSync", + connectionId: job.connectionId, + status: "failed", + daysTouched: result.daysTouched.length, + bytesWritten: result.bytesWritten, + cursorAdvanced: false, + gapDetected: false, + truncated: result.truncated === true, + errorCode: "MAIL_QUOTA_EXCEEDED", + }; + } + + /* + * A WALK THAT RAN OUT OF PAGES IS NOT A FINISHED SYNC. + * + * `history.list` hands back the mailbox's *current* head on every page, so + * a truncated walk that stored it would say "caught up" while holding only + * the first pages — and everything behind them would be skipped forever, + * with no gap signalled and the row reading `active`. `runIncrementalSync` + * reports the truncation and offers the last record it actually walked + * instead; the cursor moves there, and `catchUp` keeps this connection due + * so the next pass drains further rather than waiting out its interval. + */ + const truncated = result.truncated === true; + await ctx.runMutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId: job.connectionId, + status: "synced", + historyId: result.historyId, + daysTouched: result.daysTouched.length, + bytesWritten: result.bytesWritten, + catchUp: truncated, + }); + return { + kind: "googleForwardSync", + connectionId: job.connectionId, + status: "synced", + daysTouched: result.daysTouched.length, + bytesWritten: result.bytesWritten, + cursorAdvanced: result.historyId !== undefined, + gapDetected: false, + truncated, + }; + } catch (error) { + const { code, message } = classifyForwardSyncError(error); + // Structured, and carrying no mail: an identifier, a code, and the name of + // whatever was thrown. + console.log( + JSON.stringify({ + event: "google.forward_sync_failed", + connectionId: job.connectionId, + product: job.product, + errorCode: code, + errorName: error instanceof Error ? error.name : typeof error, + gmailStatus: error instanceof GmailApiError ? error.status : undefined, + }), + ); + return await failForwardSync(ctx, job.connectionId, code, message); + } +} + async function runGoogleGmailBackfill( ctx: ActionCtx, workspaceId: Id<"workspaces">, diff --git a/apps/convex/functions/googleConnect.ts b/apps/convex/functions/googleConnect.ts index 28f215d36..e1cacc625 100644 --- a/apps/convex/functions/googleConnect.ts +++ b/apps/convex/functions/googleConnect.ts @@ -64,6 +64,12 @@ import { hashToken } from "./lib/crypto"; import { encryptSecret, decryptSecret, requireKeyset } from "./lib/crypto"; import { randomOpaqueToken } from "./lib/gatewayAuth"; import { recordAudit } from "./lib/audit"; +// The scheduling half of a connection is the loop's (`googleSync.ts`), and the +// console reads both halves off one row. The view is imported from the leaf +// both files share rather than from the loop itself: importing the loop here +// would close a cycle, and a cycle in this module graph surfaces as an export +// that is sometimes missing rather than as an error. +import { syncStatusOf } from "./lib/googleSchedule"; import { createPkcePair, exchangeGoogleCode, @@ -217,7 +223,14 @@ function validateBackfillDays(value: number | undefined): number { return days; } -function defaultGoogleDestinationFolder( +/** + * Where a product's daily notes land when nobody has chosen a folder. + * + * Exported for `googleSync.ts`, which needs the same answer when it hands a + * pass its destination — one implementation, so a synced day and the console's + * own "daily file pattern" can never name two different folders. + */ +export function defaultGoogleDestinationFolder( service: GoogleSyncService, mailboxSlug: string | undefined, ): string { @@ -768,6 +781,29 @@ export const listGoogleConnections = query({ syncStatus: v.string(), lastSyncStartedAt: v.optional(v.number()), lastSyncCompletedAt: v.optional(v.number()), + /** + * THE ANSWER TO "IS THIS THING ACTUALLY RUNNING?" + * + * `syncStatus` above is the connection's health, which a freshly + * connected account and a happily syncing one both report as fine — + * which is exactly how a mailbox that never synced once looked identical + * to one working. These five fields are the difference: how often it is + * polled, whether it has *ever* read mail, when it is next due, and what + * went wrong last, kept after a later pass succeeded. + */ + sync: v.object({ + intervalMinutes: v.number(), + everSynced: v.boolean(), + /** A cursor exists, so new mail will be read — which is not the same as having read any. */ + cursorReady: v.boolean(), + /** The last pass ran out of history pages and there is more to drain. */ + catchingUp: v.boolean(), + lastAttemptAt: v.optional(v.number()), + nextDueAt: v.optional(v.number()), + lastFailureAt: v.optional(v.number()), + lastFailureCode: v.optional(v.string()), + lastFailure: v.optional(v.string()), + }), errorCode: v.optional(v.string()), lastError: v.optional(v.string()), gmail: v.optional( @@ -870,6 +906,7 @@ export const listGoogleConnections = query({ out.push({ connectionId: row._id, email: row.address, + sync: syncStatusOf(row), syncServices: { gmail: gmail !== undefined, calendar: calendar !== undefined, diff --git a/apps/convex/functions/googleSync.ts b/apps/convex/functions/googleSync.ts new file mode 100644 index 000000000..41ea52da6 --- /dev/null +++ b/apps/convex/functions/googleSync.ts @@ -0,0 +1,536 @@ +/** + * THE FORWARD SYNC LOOP for a connected Google account. + * + * Connecting a mailbox wrote a row, a grant and a `historyId` baseline + * (`googleConnect.ts`), and until this module existed **nothing ever advanced + * that cursor**. There was no cron entry, the gateway's `scheduled()` handler + * has no trigger configured and only ever reached the single-tenant legacy + * path anyway, and `apps/mcp/src/communications/gmailSync.js` — a complete, + * fixture-tested Gmail pipeline — was imported by nothing at all. A person + * connected Gmail in settings and their mail never arrived, with the console + * showing a connection that looked exactly like one working perfectly. + * + * So this is a **trigger, not a pipeline**: everything here decides *when* a + * connection is polled and *what the row says afterwards*. The fetching and + * the rendering are `gmailSync.js`'s, unchanged, driven from inside + * `runFileOperation` — the one credential barrier — exactly as the historical + * backfill drove them before #388 removed it. + * + * ## Why a pull, and not a webhook + * + * Google can push: `users.watch` posts Gmail changes to a Pub/Sub topic, and + * Calendar has watch channels. Both are rejected here for now, and + * `docs/decisions/communications.md` carries the argument in full — the short + * version is that a push path needs a Google Cloud Pub/Sub topic in *our* + * project, an internet-facing verified endpoint, and a re-registration every + * seven days per mailbox, and it still needs this loop underneath it, because + * a missed or expired watch is only ever noticed by something that polls. A + * pull that works is worth more than a push that needs a pull to be correct. + * + * ## Why the floor is five minutes, and how one cron serves many intervals + * + * `crons.ts` runs `sweepDueGoogleSyncs` every five minutes — one job, at the + * floor — and the sweep starts a pass only for connections that are actually + * due (`now >= lastSyncAt + interval`). A per-connection + * `syncIntervalMinutes` is therefore served by a fixed global tick: the cron + * decides when to *look*, never whether a given connection may sync. + * + * Five minutes is the floor because it is the resolution of the tick, and + * because below it the poll stops being cheap: every pass mints or reuses an + * access token, calls `history.list`, and re-lists and re-renders every day a + * changed message landed on. `apps/desktop/src/main/imessage.ts` — the sync + * loop in this codebase that actually works — settled on the same five + * minutes against a local SQLite file, and this one talks to Google over the + * network on somebody else's quota. + * + * The **default** is fifteen minutes rather than the floor. Mail is not a + * chat: three passes an hour keeps a brain within a quarter of an hour of the + * mailbox, at a third of the floor's cost in Convex actions and Google calls, + * and the person who wants the floor can set it. What a person loses by + * choosing a longer interval is written down in + * `docs/decisions/communications.md` and is worth stating plainly: the mail is + * not lost, it is late — every pass rebuilds each touched day from Gmail's + * live state, so a slower poll writes the same bytes later. The one real risk + * is at the far end: Gmail expires a `historyId` after roughly a week, so an + * interval measured in days makes an expired cursor likely, and an expired + * cursor under a forward-only policy means the mail that arrived in the gap is + * never captured. That is why `MAX_SYNC_INTERVAL_MINUTES` is one day. + * + * ## Forward-only + * + * #388 disabled historical backfill deliberately and that decision stands. + * Nothing here re-enables it and nothing here routes around + * `GOOGLE_SYNC_FORWARD_ONLY`: a pass advances `historyId` from wherever it is, + * which is what the cursor was recorded for. A connection with no cursor yet + * takes one from `users.getProfile` and starts there — the moment it is + * adopted, not a day earlier. + */ + +import { ConvexError, v } from "convex/values"; +import { internalMutation, internalQuery, mutation } from "../_generated/server"; +import { internal } from "../_generated/api"; +import { recordAudit } from "./lib/audit"; +import { defaultGoogleDestinationFolder, mailConnectEnabled, requireActor } from "./googleConnect"; +/* + * The arithmetic lives in a leaf module rather than here, because + * `googleConnect.ts` reads it too and a cycle between these two files fails as + * an intermittently missing export rather than as an error — see that file's + * own comment for the shape it took: a test about deleting an account, failing + * one full-suite run in six, naming a function neither file mentions. + */ +import { + DEFAULT_SYNC_INTERVAL_MINUTES, + MAX_SYNC_INTERVAL_MINUTES, + MIN_SYNC_INTERVAL_MINUTES, + SYNC_STALL_MS, + failureBackoffMs, + isDue, + syncIntervalMinutesOf, + syncableProductsOf, +} from "./lib/googleSchedule"; + +/** Connections one sweep may start, bounded like every other sweep here. */ +const SWEEP_BATCH = 50; + +function validateSyncInterval(value: number): number { + if (!Number.isFinite(value) || !Number.isInteger(value)) { + throw new ConvexError({ + code: "GOOGLE_SYNC_INTERVAL_INVALID", + message: `Choose a whole number of minutes, at least ${MIN_SYNC_INTERVAL_MINUTES}.`, + }); + } + if (value < MIN_SYNC_INTERVAL_MINUTES) { + throw new ConvexError({ + code: "GOOGLE_SYNC_INTERVAL_TOO_SHORT", + message: `Sync can run at most every ${MIN_SYNC_INTERVAL_MINUTES} minutes.`, + }); + } + if (value > MAX_SYNC_INTERVAL_MINUTES) { + throw new ConvexError({ + code: "GOOGLE_SYNC_INTERVAL_TOO_LONG", + message: `Sync must run at least once a day; ${MAX_SYNC_INTERVAL_MINUTES} minutes is the longest gap.`, + }); + } + return value; +} + +/** + * How often this account is polled. **Owner of a personal context only.** + * + * The floor lives here rather than in the picker, because a picker is a + * suggestion: this refuses four minutes from a console, from a script, and + * from a client that has never seen the UI. + */ +export const updateGoogleSyncInterval = mutation({ + args: { + workspaceId: v.id("workspaces"), + connectionId: v.id("googleConnections"), + syncIntervalMinutes: v.number(), + }, + returns: v.null(), + handler: async (ctx, args) => { + const userId = await requireActor(ctx); + const allowed = await ctx.runQuery(internal.functions.googleConnect.requirePersonalOwner, { + workspaceId: args.workspaceId, + userId, + }); + if (!allowed) { + throw new ConvexError({ + code: "NOT_OWNER", + message: "Only the owner can change how often this context syncs.", + }); + } + const minutes = validateSyncInterval(args.syncIntervalMinutes); + const connection = await ctx.db.get(args.connectionId); + /* + One refusal for every way of saying no, and `workspaceId` is checked + against the row rather than trusted from the caller. An owner of context + A naming a connection that belongs to context B must not be able to tell + "that is not yours" apart from "there is no such connection" — the same + rule `updateGoogleSyncDestination` applies one file over. + */ + if ( + connection === null || + connection.workspaceId !== args.workspaceId || + connection.disconnectedAt !== undefined + ) { + throw new ConvexError({ + code: "GOOGLE_CONNECTION_NOT_FOUND", + message: "That Google account is not connected to this context.", + }); + } + + await ctx.db.patch(args.connectionId, { + syncIntervalMinutes: minutes, + /* + A shortened interval takes effect at once rather than after one more + wait at the old one — the same property `imessage.ts` gets by reading + its settings fresh every pass. A connection that has never synced keeps + no `nextSyncAt` at all, because it is already due. + */ + nextSyncAt: + connection.lastSyncAt === undefined ? undefined : connection.lastSyncAt + minutes * 60_000, + updatedAt: Date.now(), + }); + await recordAudit(ctx, { + workspaceId: args.workspaceId, + actorUserId: userId, + action: "google_sync_interval_updated", + details: { connectionId: args.connectionId, syncIntervalMinutes: minutes }, + }); + return null; + }, +}); + +/** + * THE CRON'S ONLY JOB: look, and start passes for connections that are due. + * + * The argument `crons.ts` requires of anything that starts work rather than + * deleting it, made here so it can be pointed at: **this holds no decision.** + * Whether a connection may sync at all is the connection's own state — is it + * still connected, is it a personal context, does it enable a product this + * engine can advance, does this deployment allow reading a restricted scope — + * and every one of those is re-asked by `googleForwardSyncJob` inside the pass + * itself, before a credential is opened. What the sweep decides is only *when + * to look*, and the answer is "every five minutes, at whichever connections + * their own interval has made due". + * + * The deployment flag is the one thing read here as well as there, and it is + * read as a switch rather than as a decision: with Google's verification not + * yet granted, a deployment that may not read mail should not be starting + * scheduled work that opens a credential to discover that. + */ +export const sweepDueGoogleSyncs = internalMutation({ + args: {}, + returns: v.object({ started: v.number(), examined: v.number() }), + handler: async (ctx): Promise<{ started: number; examined: number }> => { + if (!mailConnectEnabled()) return { started: 0, examined: 0 }; + const now = Date.now(); + const rows = await ctx.db + .query("googleConnections") + .withIndex("by_sync_due", (q) => q.eq("disconnectedAt", undefined).lte("nextSyncAt", now)) + // Bounded, like every other sweep here: a backlog drains over several + // runs rather than in one transaction big enough to hit a limit. The + // index is ordered by due time, so the oldest-due drain first and no + // connection can be starved by one that is not due yet. + .take(SWEEP_BATCH); + + let started = 0; + for (const row of rows) { + // The index range already excludes these. An index is a poor place to + // trust an invariant that lives on another field — the same belt + // `sweepStalledBackfills` wears over `optedIn`. Unreachable while the + // index above is right, which is the point: it is what catches the day + // somebody changes the index. + if (row.disconnectedAt !== undefined) continue; + if (syncableProductsOf(row).length === 0) continue; + if (!isDue(row, now)) continue; + /* + NOT OVERTAKING A PASS THAT IS STILL RUNNING. + + A claimed pass sets `syncStartedAt` and clears it when it reports, so a + pass in flight looks recent and a lost one looks stale. Two passes on + one mailbox is not a correctness failure — every day is rebuilt from + Gmail's live state, so the second would write the same bytes — but it + doubles the Google calls and races two writers onto one note's etag, + and there is no reason to cause it on purpose. + */ + if (row.syncStartedAt !== undefined && now - row.syncStartedAt < SYNC_STALL_MS) continue; + + const intervalMs = syncIntervalMinutesOf(row) * 60_000; + await ctx.db.patch(row._id, { + syncStartedAt: now, + // Claimed, so the next tick finds it not due even if this pass never + // reports. A pass that is genuinely lost is picked up again by the + // stall guard above. + nextSyncAt: now + intervalMs, + updatedAt: now, + }); + await ctx.scheduler.runAfter(0, internal.functions.files.runFileOperation, { + workspaceId: row.workspaceId, + scope: "private", + operation: { kind: "googleForwardSync", connectionId: row._id }, + }); + started += 1; + } + return { started, examined: rows.length }; + }, +}); + +const forwardSyncJobValidator = v.union( + v.null(), + v.object({ kind: v.literal("skip"), reason: v.string() }), + v.object({ + kind: v.literal("run"), + connectionId: v.id("googleConnections"), + product: v.literal("gmail"), + address: v.string(), + mailboxSlug: v.string(), + destinationFolder: v.string(), + folders: v.array(v.union(v.literal("inbox"), v.literal("sent"))), + quotaBytes: v.number(), + bytesAlreadyUsed: v.number(), + attachmentMode: v.union(v.literal("metadata-only"), v.literal("store")), + attachmentRetentionDays: v.optional(v.union(v.number(), v.literal("forever"))), + historyId: v.optional(v.string()), + }), +); + +/** + * What the pass is allowed to do, re-asked at the moment it runs. + * + * The sweep that scheduled this ran up to a few minutes ago and holds no + * decision (see above); in that time the account can have been disconnected, + * the product turned off, or the grant revoked. So every gate is asked here, + * **before** `runFileOperation` opens a bucket credential — the same ordering, + * and the same reason, as the projection pass that asks + * `projectionTargetForWorkspace` first. + * + * `null` means there is no row to report against at all. `skip` means there is + * a row, it is not to be synced, and the claim on it must still be released. + * + * No secret is returned. This is the settings-and-cursor view; the refresh + * token never leaves `mintGoogleAccessToken`. + */ +export const googleForwardSyncJob = internalQuery({ + args: { workspaceId: v.id("workspaces"), connectionId: v.id("googleConnections") }, + returns: forwardSyncJobValidator, + handler: async (ctx, args) => { + const connection = await ctx.db.get(args.connectionId); + if (connection === null) return null; + /* + * THE TWO ARGUMENTS HAVE TO AGREE, AND THIS IS WHERE THAT IS CHECKED. + * + * The pass runs inside `runFileOperation`, which opens the bucket + * credential of the `workspaceId` it was handed, and then writes mail read + * from the account named by `connectionId`. A mismatched pair would file + * one context's mail into another context's bucket. Nothing constructs + * such a pair today — the sweep reads both off one row — but "no caller + * does that" is not a tenant boundary, and this is the one place that can + * be one, because it is the only function that sees both. + */ + if (connection.workspaceId !== args.workspaceId) return null; + if (connection.disconnectedAt !== undefined) { + return { kind: "skip" as const, reason: "GOOGLE_DISCONNECTED" }; + } + if (!mailConnectEnabled()) { + return { kind: "skip" as const, reason: "MAIL_CONNECT_DISABLED" }; + } + const workspace = await ctx.db.get(args.workspaceId); + if (workspace === null || workspace.kind !== "personal") { + // A mailbox may only ever land in a personal context + // (`identity-and-access.md`). A workspace that changed kind under a + // connection is not a state this writes mail through. + return { kind: "skip" as const, reason: "NOT_PERSONAL_CONTEXT" }; + } + if (connection.health === "reconnect_required") { + // Minting would fail against a grant Google has already refused, once + // per pass, forever. A reconnect rewrites `health`, which is what puts + // this connection back in the loop. + return { kind: "skip" as const, reason: "GOOGLE_RECONNECT_REQUIRED" }; + } + if (!connection.products.includes("gmail") || !connection.gmail) { + return { kind: "skip" as const, reason: "NO_SYNCABLE_PRODUCT" }; + } + + const gmail = connection.gmail; + return { + kind: "run" as const, + connectionId: connection._id, + product: "gmail" as const, + address: connection.address, + mailboxSlug: gmail.mailboxSlug, + destinationFolder: + gmail.destinationFolder ?? defaultGoogleDestinationFolder("gmail", gmail.mailboxSlug), + folders: gmail.folders, + quotaBytes: gmail.quotaBytes, + bytesAlreadyUsed: connection.syncBytesWritten ?? 0, + attachmentMode: gmail.attachmentMode, + attachmentRetentionDays: gmail.attachmentRetentionDays, + historyId: gmail.historyId, + }; + }, +}); + +/** + * What one pass learned, written where a person can see it. + * + * Every exit from a pass comes through here, including the ones that did + * nothing: the claim (`syncStartedAt`) has to be released or the connection + * sits unsyncable for fifteen minutes for no reason. + * + * `lastSyncAt` is when a pass last **finished**, successfully or not, and the + * next due time is computed from it. That is deliberate: making it mean "last + * success" would leave a permanently failing connection due on every single + * tick, which is the request pattern most likely to keep it failing. + * "Last synced", the thing a person actually asks about, is + * `gmail.lastSyncedAt`, and it moves only when mail was read. + */ +export const recordGoogleForwardSyncPass = internalMutation({ + args: { + connectionId: v.id("googleConnections"), + status: v.union(v.literal("synced"), v.literal("skipped"), v.literal("failed")), + /** The cursor to store. Absent leaves the existing one exactly where it is. */ + historyId: v.optional(v.string()), + daysTouched: v.optional(v.number()), + bytesWritten: v.optional(v.number()), + /** Gmail expired the cursor. Forward-only: it is re-baselined and the gap is recorded, not backfilled. */ + gapDetected: v.optional(v.boolean()), + /** + * The history walk ran out of pages. The cursor still moves — to the last + * record walked — and the connection stays due, so the next pass drains + * further instead of waiting out an interval it already knows is wrong. + */ + catchUp: v.optional(v.boolean()), + /** + * This pass established a cursor rather than reading mail: a first + * baseline, or a re-baseline after a gap. It is a successful pass and it + * read nothing, so `gmail.lastSyncedAt` — the console's "has this ever + * actually synced" — deliberately does not move. + */ + baseline: v.optional(v.boolean()), + errorCode: v.optional(v.string()), + error: v.optional(v.string()), + }, + returns: v.object({ accepted: v.boolean() }), + handler: async (ctx, args): Promise<{ accepted: boolean }> => { + const connection = await ctx.db.get(args.connectionId); + if (connection === null || connection.disconnectedAt !== undefined) { + return { accepted: false }; + } + const now = Date.now(); + const intervalMs = syncIntervalMinutesOf(connection) * 60_000; + const error = args.error?.slice(0, 240); + + if (args.status === "skipped") { + // Nothing was read, so `lastSyncAt` does not move — a skipped pass must + // not be able to make a connection that has never synced look synced. + await ctx.db.patch(args.connectionId, { + syncStartedAt: undefined, + nextSyncAt: now + intervalMs, + updatedAt: now, + }); + return { accepted: true }; + } + + if (args.status === "failed") { + const failures = (connection.syncFailures ?? 0) + 1; + await ctx.db.patch(args.connectionId, { + syncStartedAt: undefined, + lastSyncAt: now, + nextSyncAt: now + failureBackoffMs(intervalMs, failures, args.connectionId), + syncFailures: failures, + /* + BYTES ARE COUNTED ON THE PATH THAT ACTUALLY WRITES THEM. + + The one caller that reports a non-zero figure here is the quota path, + which stops *after* writing whole days. Dropping the count froze + `bytesAlreadyUsed` below the ceiling, so every later pass re-listed, + re-rendered and re-wrote the same days and dropped the bytes again — + a ceiling that could never be crossed, which is what the schema + comment calls decoration. + */ + syncBytesWritten: (connection.syncBytesWritten ?? 0) + (args.bytesWritten ?? 0), + /* + `reconnect_required` SURVIVES A FAILED PASS. + + `mintGoogleAccessToken` sets it when Google refuses the grant + outright, and the pass's own skip gate keys on it — so overwriting it + with a plain `error` here meant the gate never fired again and a dead + grant was offered to Google's token endpoint every backoff, forever. + It also erased the one state the console renders as "needs + reconnect", which is the only thing the owner can act on. + */ + health: + connection.health === "reconnect_required" + ? ("reconnect_required" as const) + : ("error" as const), + lastError: error ?? "This Google account did not sync.", + errorCode: args.errorCode ?? "GOOGLE_SYNC_FAILED", + lastSyncFailureAt: now, + lastSyncFailureCode: args.errorCode ?? "GOOGLE_SYNC_FAILED", + lastSyncFailure: error ?? "This Google account did not sync.", + updatedAt: now, + }); + return { accepted: true }; + } + + const gmail = connection.gmail + ? { + ...connection.gmail, + historyId: args.historyId ?? connection.gmail.historyId, + // A baseline or a re-baseline read no mail, so it does not claim to + // have. `cursorReady` is what those passes make true. + lastSyncedAt: args.baseline === true ? connection.gmail.lastSyncedAt : now, + } + : connection.gmail; + /* + A GAP IS RECORDED, NOT SMOOTHED OVER. + + Gmail dropped this mailbox's `historyId`, so the changes since it was + issued cannot be enumerated. The documented recovery was a full reconcile + over the backfill window, and forward-only (#388) does not have one — so + the cursor is re-baselined to Gmail's current one and the mail that + arrived in the gap is not captured. That is a real loss and it is written + onto the row as a failure a person can read, even though the pass itself + succeeded and the connection is healthy from here on. + */ + const gap = args.gapDetected === true; + const catchUp = args.catchUp === true; + await ctx.db.patch(args.connectionId, { + gmail, + syncStartedAt: undefined, + lastSyncAt: now, + // A pass that knows it left work behind is due at once; anything else + // waits its interval. `syncCatchUp` is cleared either way, so a + // connection that has caught up stops being due every tick. + nextSyncAt: catchUp ? now : now + intervalMs, + syncCatchUp: catchUp ? true : undefined, + syncFailures: undefined, + syncBytesWritten: (connection.syncBytesWritten ?? 0) + (args.bytesWritten ?? 0), + health: "active" as const, + lastError: undefined, + errorCode: undefined, + ...(gap + ? { + lastSyncFailureAt: now, + lastSyncFailureCode: "GOOGLE_SYNC_GAP", + lastSyncFailure: + "Google expired this mailbox's sync cursor. Mail that arrived while it was expired was not captured; syncing continues from now.", + } + : {}), + updatedAt: now, + }); + + /* + A PASS THAT WROTE INTO SOMEBODY'S BUCKET LEAVES A ROW SAYING SO. + + This is the first writer in the codebase with no person present, and + non-negotiable #4 says the audit records the acting identity rather than + just the scope. The identity here is `boundBy` — whoever connected the + account, on whose grant every one of these writes is made. Naming them is + more honest than an empty actor, and it is the name an owner needs when + the question is "who set this up". + + Only passes that actually wrote something. A poll that found nothing is + not an event, and recording 288 of those a day per connection would bury + the ones that matter. Counts only: no subject, no address, no path — the + day notes' own paths are `gmail.destinationFolder` plus a date, which the + row already carries. + */ + if ((args.daysTouched ?? 0) > 0) { + await recordAudit(ctx, { + workspaceId: connection.workspaceId, + actorUserId: connection.boundBy, + action: "google_sync_wrote", + details: { + connectionId: args.connectionId, + product: "gmail", + days: args.daysTouched ?? 0, + bytes: args.bytesWritten ?? 0, + }, + }); + } + return { accepted: true }; + }, +}); diff --git a/apps/convex/functions/lib/googleSchedule.ts b/apps/convex/functions/lib/googleSchedule.ts new file mode 100644 index 000000000..90fc4502a --- /dev/null +++ b/apps/convex/functions/lib/googleSchedule.ts @@ -0,0 +1,192 @@ +/** + * The scheduling arithmetic behind the Google forward sync loop. + * + * A LEAF: this module imports nothing from `functions/`, and that is the whole + * reason it exists. `functions/googleSync.ts` owns the loop and needs the + * connect module's helpers; `functions/googleConnect.ts` owns the console's + * listing and needs the loop's status view. Putting the shared half in either + * of them makes the two import each other, and a cycle in the Convex module + * graph does not fail loudly — it fails as an export that is *sometimes* + * missing, depending on which module the loader happened to enter first. That + * is not a hypothetical: it showed up as `revokeGoogleGrant` resolving to "no + * such export" in roughly one full-suite run in six, in a test about deleting + * an account, which has nothing to do with either file. + * + * Everything here is a pure function of a row. See `functions/googleSync.ts` + * for what the numbers mean and `docs/decisions/communications.md` for why + * they are what they are. + */ + +import type { Doc } from "../../_generated/dataModel"; + +/** + * The lowest interval this deployment will accept, in minutes — **refused + * server-side**, not merely absent from a picker. A client that posts 1 is + * refused with the same error as one that posts 0 or 4.5. + */ +export const MIN_SYNC_INTERVAL_MINUTES = 5; + +/** The interval a connection has when its owner has never chosen one. */ +export const DEFAULT_SYNC_INTERVAL_MINUTES = 15; + +/** + * The longest interval, one day. Not a policy about attention — a bound on + * cursor expiry: Gmail drops a `historyId` after about a week, and under a + * forward-only policy an expired cursor is mail nobody ever fetches. + */ +export const MAX_SYNC_INTERVAL_MINUTES = 24 * 60; + +/** + * How long a claimed pass may be silent before another may start. + * + * The same fifteen minutes, and the same argument, as + * `fastSearch.sweepStalledBackfills`: a pass that is still running must never + * be overtaken by a second one, and the only evidence a mutation has of a + * running action is that something wrote to the row recently. `syncStartedAt` + * is that heartbeat — set when the pass is claimed, cleared when it reports. + */ +export const SYNC_STALL_MS = 15 * 60 * 1000; + +/** + * The floor on how soon a *failed* pass is retried, whatever the interval. + * + * A connection Google is rate-limiting, or one whose grant has been revoked, + * would otherwise be retried every five minutes forever by whoever chose the + * floor — which is the request pattern most likely to keep it rate-limited. + */ +export const SYNC_FAILURE_BACKOFF_MS = 15 * 60 * 1000; + +/** + * The ceiling on that backoff: six hours. + * + * A ladder with no cap turns a fortnight of failures into a connection nobody + * ever checks again, and the failures this actually meets — a revoked grant, a + * daily quota, a bucket somebody has to reconnect — are all fixed by a person + * doing something, after which the next pass should be hours away rather than + * days. + */ +export const MAX_SYNC_BACKOFF_MS = 6 * 60 * 60 * 1000; + +/** + * How long to wait after a failure: the interval, or the ladder, whichever is + * longer — plus a per-connection spread so a deployment's connections do not + * all wake in the same minute after a Google outage ends. + * + * The spread is derived from the row id rather than drawn at random, because + * this is computed inside a mutation and a value a test cannot predict is a + * value a test cannot pin. + */ +export function failureBackoffMs( + intervalMs: number, + failures: number, + connectionId: string, +): number { + const step = Math.max(0, Math.min(Math.floor(failures) - 1, 8)); + const ladder = Math.min(MAX_SYNC_BACKOFF_MS, SYNC_FAILURE_BACKOFF_MS * 2 ** step); + return Math.max(intervalMs, ladder) + spreadMs(connectionId); +} + +/** Up to a minute, stable for one connection, different between connections. */ +function spreadMs(connectionId: string): number { + let hash = 0; + for (let index = 0; index < connectionId.length; index += 1) { + hash = (hash * 31 + connectionId.charCodeAt(index)) % 60_000; + } + return hash; +} + +/** + * Which products the loop can actually advance today. + * + * Gmail, and the loop is deliberately built around the *account* rather than + * around Gmail: one row, one grant, one claim, one report. Calendar and Chat + * join by being added here and given a pass in `functions/files.ts` — they do + * not need a second cron, a second claim, or a second set of status fields. + * See `docs/decisions/communications.md` for what each of them still needs. + */ +export const ENGINE_PRODUCTS = ["gmail"] as const; + +/** The interval in force for a row: the owner's choice, or the default, never below the floor. */ +export function syncIntervalMinutesOf(connection: { syncIntervalMinutes?: number }): number { + const chosen = connection.syncIntervalMinutes; + if (typeof chosen !== "number" || !Number.isFinite(chosen)) { + return DEFAULT_SYNC_INTERVAL_MINUTES; + } + return Math.max(MIN_SYNC_INTERVAL_MINUTES, Math.floor(chosen)); +} + +/** Products on this row the loop can sync. Empty means there is nothing to poll for. */ +export function syncableProductsOf(connection: { products: string[] }): string[] { + return ENGINE_PRODUCTS.filter((product) => connection.products.includes(product)); +} + +/** + * Is this connection due, by the contract the cron is written against: + * `now >= lastSyncAt + interval`. + * + * Derived from `lastSyncAt` rather than read from `nextSyncAt`, on purpose. + * `nextSyncAt` exists so the sweep can ask an index for candidates instead of + * reading every connection in the deployment; it is a materialized copy, and a + * copy is the thing that can be stale. The answer comes from the two facts it + * was computed from. + */ +export function isDue( + connection: { lastSyncAt?: number; syncIntervalMinutes?: number; syncCatchUp?: boolean }, + now: number, +): boolean { + /* + A pass that ran out of pages is due again at once, whatever the interval. + The interval is how often to *ask whether anything changed*; this + connection is not asking, it is draining a backlog it has already seen the + edge of, and every pass makes real progress because the cursor moved to the + last record walked. + */ + if (connection.syncCatchUp === true) return true; + if (connection.lastSyncAt === undefined) return true; + return now >= connection.lastSyncAt + syncIntervalMinutesOf(connection) * 60_000; +} + +/** The schedule as the console reads it. One implementation, two readers. */ +export function syncStatusOf(connection: Doc<"googleConnections">): { + intervalMinutes: number; + everSynced: boolean; + cursorReady: boolean; + catchingUp: boolean; + lastAttemptAt?: number; + nextDueAt?: number; + lastFailureAt?: number; + lastFailureCode?: string; + lastFailure?: string; +} { + const intervalMinutes = syncIntervalMinutesOf(connection); + return { + intervalMinutes, + /* + The distinction the product has been missing. A connection that has never + synced and one syncing fine were the same screen, and this is the field + that separates them: it is about mail actually read, so a pass that was + skipped or that failed does not make it true. + */ + everSynced: connection.gmail?.lastSyncedAt !== undefined, + /* + A cursor exists, so this connection is watching — but watching is not the + same as having read anything, and the two used to be one screen. A + baseline pass (and a re-baseline after a gap) sets this and deliberately + leaves `everSynced` alone, because forward-only means neither one read a + single message. + */ + cursorReady: connection.gmail?.historyId !== undefined, + catchingUp: connection.syncCatchUp === true, + lastAttemptAt: connection.lastSyncAt, + nextDueAt: + connection.disconnectedAt !== undefined + ? undefined + : (connection.nextSyncAt ?? + (connection.lastSyncAt === undefined + ? undefined + : connection.lastSyncAt + intervalMinutes * 60_000)), + lastFailureAt: connection.lastSyncFailureAt, + lastFailureCode: connection.lastSyncFailureCode, + lastFailure: connection.lastSyncFailure, + }; +} diff --git a/apps/convex/schema.ts b/apps/convex/schema.ts index 9c07d7ebd..8d1b4e9aa 100644 --- a/apps/convex/schema.ts +++ b/apps/convex/schema.ts @@ -1136,6 +1136,77 @@ const schema = defineSchema({ ), lastError: v.optional(v.string()), errorCode: v.optional(v.string()), + /** + * HOW OFTEN THIS ACCOUNT IS POLLED, AND WHEN IT IS NEXT DUE. + * + * The fields below are the whole scheduling state of the forward sync + * loop (`functions/googleSync.ts`), and they are **per account, not per + * product**: one Google account is one grant, so one pass mints one + * access token and walks whichever products the row enables. A per-product + * schedule would mint the same credential three times an hour to ask three + * questions of the same account. + * + * They sit at the top level rather than inside `gmail` for that reason and + * for one more: `nextSyncAt` is indexed, and an index over a field nested + * inside an optional object is a shape this schema does not otherwise use. + * + * - `syncIntervalMinutes` — the owner's choice, floored at + * `MIN_SYNC_INTERVAL_MINUTES` server-side. Absent means the default + * (`DEFAULT_SYNC_INTERVAL_MINUTES`), so a row written before this + * existed is scheduled rather than stalled. + * - `lastSyncAt` — when a pass last **finished**, successfully or not. + * Absent means this connection has never synced, which the console + * must be able to say out loud: "connected" and "syncing" looking + * identical is the defect this loop exists to close. + * - `nextSyncAt` — `lastSyncAt + interval`, materialized so the sweep can + * ask the index for due rows instead of reading every connection. + * Absent means due now, which is what a never-synced row is. + * - `syncStartedAt` — set when a pass is claimed, cleared when it + * reports. It is the not-overtaking guard: a pass still running is + * never started a second time until it has been silent long enough to + * be considered lost. + * - `lastSyncFailure*` — the last failure this connection had, kept + * **after** a later pass succeeds. `lastError` / `errorCode` describe + * the connection's health right now and are cleared by a good pass; + * somebody asking "did this break overnight?" is asking a different + * question, and clearing the answer is how it stopped being askable. + */ + syncIntervalMinutes: v.optional(v.number()), + lastSyncAt: v.optional(v.number()), + nextSyncAt: v.optional(v.number()), + syncStartedAt: v.optional(v.number()), + lastSyncFailureAt: v.optional(v.number()), + lastSyncFailureCode: v.optional(v.string()), + lastSyncFailure: v.optional(v.string()), + /** + * The last pass ran out of history pages before it ran out of history. + * + * Gmail's `history.list` is paged and the walk is bounded, so a connection + * whose cursor is weeks old cannot be caught up in one pass. The cursor + * still moves — to the last record actually walked, never to the mailbox + * head — and this says the interval must not be waited out, because the + * pass already knows there is more. `isDue` reads it; a pass that finishes + * clears it, which is what stops a connection being due forever. + */ + syncCatchUp: v.optional(v.boolean()), + /** + * Consecutive failed passes, cleared by the first good one. + * + * The backoff ladder's input. A flat retry means a mailbox Google is + * rate-limiting is asked again ~96 times a day, which is the request + * pattern most likely to keep it rate-limited. + */ + syncFailures: v.optional(v.number()), + /** + * Bytes the forward loop has written into the bucket for this connection. + * + * `gmail.quotaBytes` is a lifetime ceiling on what one connection may + * write, and a ceiling with nothing counting against it is decoration. The + * historical backfill counted on its run row; a forward loop has no run, + * so the total lives here and every pass is handed it as + * `bytesAlreadyUsed`. + */ + syncBytesWritten: v.optional(v.number()), /** * Set by disconnect. The row is kept — never deleted outright — so a * disconnected connection's sync job can be told apart from one that @@ -1175,7 +1246,20 @@ const schema = defineSchema({ }) .index("by_workspace", ["workspaceId"]) /** One connection per address per context — the uniqueness `chooseMailboxSlug` assumes for Gmail. */ - .index("by_workspace_address", ["workspaceId", "address"]), + .index("by_workspace_address", ["workspaceId", "address"]) + /** + * The sweep's index: connections that are still connected, oldest due + * first. + * + * `disconnectedAt` leads so a disconnected row is outside the range + * entirely rather than filtered out after being read — a disconnected + * connection with an old `nextSyncAt` would otherwise sit at the head of + * every bounded batch forever and starve the live ones behind it. + * + * A row with no `nextSyncAt` sorts before every number, so a never-synced + * connection is at the front of the queue rather than invisible to it. + */ + .index("by_sync_due", ["disconnectedAt", "nextSyncAt"]), /** * User-visible Google sync work. diff --git a/apps/mcp/src/communications/gmailSync.js b/apps/mcp/src/communications/gmailSync.js index 859f45646..1b8bb2e36 100644 --- a/apps/mcp/src/communications/gmailSync.js +++ b/apps/mcp/src/communications/gmailSync.js @@ -753,27 +753,79 @@ export async function listHistoryPage({ fetchImpl, accessToken, startHistoryId, throw error; } const ids = new Set(); + let lastRecordId; for (const record of body.history ?? []) { + if (record?.id !== undefined) { + const id = String(record.id); + if (lastRecordId === undefined || historyIdIsAfter(id, lastRecordId)) lastRecordId = id; + } for (const added of record.messagesAdded ?? []) { if (added?.message?.id) ids.add(String(added.message.id)); } } - return { messageIds: ids, nextPageToken: body.nextPageToken, historyId: body.historyId }; + return { + messageIds: ids, + nextPageToken: body.nextPageToken, + historyId: body.historyId, + lastRecordId, + }; } -/** Every message id added since `startHistoryId`, and the historyId to resume from next time. */ +/** + * Is `a` a later history id than `b`? + * + * Compared as decimal digit strings — length first, then lexicographically — + * rather than through `Number`. A Gmail `historyId` is an unsigned 64-bit + * value delivered as a string, and the ones large enough to lose precision as + * a double are exactly the ones nobody would notice going wrong. + */ +function historyIdIsAfter(a, b) { + const left = String(a).replace(/^0+(?=\d)/, ""); + const right = String(b).replace(/^0+(?=\d)/, ""); + if (left.length !== right.length) return left.length > right.length; + return left > right; +} + +/** + * Every message id added since `startHistoryId`, and where to resume. + * + * **`history.list` returns the MAILBOX'S CURRENT `historyId` on every page**, + * not a per-page cursor. A walk that stops at `maxPages` and reports that + * value tells its caller "you are caught up" while holding only the first N + * pages — and everything after them is then skipped forever, silently, with no + * gap signalled. That is the one failure mode in this whole path that loses + * somebody's mail without saying so, and a mailbox whose cursor is weeks old + * is precisely where it fires. + * + * So a truncated walk says `truncated: true` and carries `lastRecordId`: the + * id of the last history *record* it actually walked, which is a valid + * `startHistoryId` for the next call and covers exactly the records collected + * here. Resuming from it is what makes a truncated pass make progress rather + * than repeat itself. `historyId` still reports the mailbox head, because a + * caller that reached the end wants it — but a caller must consult + * `truncated` before believing it. + */ export async function listAllHistory({ fetchImpl, accessToken, startHistoryId, maxPages = 50 }) { const messageIds = new Set(); let pageToken; let historyId = startHistoryId; + let lastRecordId; + let truncated = false; for (let page = 0; page < maxPages; page += 1) { const result = await listHistoryPage({ fetchImpl, accessToken, startHistoryId, pageToken }); for (const id of result.messageIds) messageIds.add(id); if (result.historyId) historyId = result.historyId; + if ( + result.lastRecordId !== undefined && + (lastRecordId === undefined || historyIdIsAfter(result.lastRecordId, lastRecordId)) + ) { + lastRecordId = result.lastRecordId; + } if (!result.nextPageToken) break; pageToken = result.nextPageToken; + if (page + 1 >= maxPages) truncated = true; } - return { messageIds, historyId }; + return { messageIds, historyId, truncated, lastRecordId }; } /* -------------------------------------------------------------------------- */ @@ -852,6 +904,21 @@ export function renderDay(options) { export async function writeDayPart(store, part, maxAttempts = 3) { const bytes = new TextEncoder().encode(part.text).length; if (!store.capabilities?.conditionalWrite) { + /* + NO CONDITIONAL WRITE STILL MEANS NO POINTLESS WRITE. + This branch used to `put` unconditionally, which made "re-syncing an + unchanged day writes nothing" true on R2 and S3 and false on exactly the + backends CLAUDE.md already flags — B2 and Wasabi — where a scheduled loop + would then rewrite every touched day on every pass forever, and count the + bytes again each time against the connection's quota. The read-compare is + the same one the conditional branch does; what this backend cannot give + is the *atomicity* that turns a race into a retry, and that is the + degradation, not "write blindly". + */ + const current = await store.get(part.path); + if (current && (await current.text()) === part.text) { + return { path: part.path, bytes, wrote: false }; + } await store.put(part.path, part.text); return { path: part.path, bytes, wrote: true }; } @@ -1085,8 +1152,15 @@ export async function runBackfill(options) { * connection's window again, which regenerates every day from live state and * is therefore a correct reconcile regardless of what was missed. * + * `truncated` is the other half of the same honesty: the history walk ran out + * of pages before it ran out of history, so `historyId` here is the last + * record walked rather than the mailbox head, and the caller has more to do. + * A caller that ignores it and stores the cursor anyway is still correct about + * what it wrote; it is only wrong about being finished — which is why this is + * returned rather than thrown. + * * @returns {Promise<{gapDetected: boolean, daysTouched: string[], bytesWritten: number, - * quotaExceeded: boolean, historyId?: string}>} + * quotaExceeded: boolean, historyId?: string, truncated: boolean}>} */ export async function runIncrementalSync(options) { let history; @@ -1095,16 +1169,30 @@ export async function runIncrementalSync(options) { fetchImpl: options.fetchImpl, accessToken: options.accessToken, startHistoryId: options.startHistoryId, + ...(options.maxHistoryPages === undefined ? {} : { maxPages: options.maxHistoryPages }), }); } catch (error) { if (error instanceof GmailHistoryExpiredError) { - return { gapDetected: true, daysTouched: [], bytesWritten: 0, quotaExceeded: false }; + return { gapDetected: true, daysTouched: [], bytesWritten: 0, quotaExceeded: false, truncated: false }; } throw error; } + // Where the next pass should start. A complete walk ends at the mailbox + // head; a truncated one ends at the last record it actually read, and + // `undefined` (a truncated walk that saw no record ids at all) means "do not + // move the cursor", which the caller must honour. + const resumeFrom = history.truncated ? history.lastRecordId : history.historyId; + if (history.messageIds.size === 0) { - return { gapDetected: false, daysTouched: [], bytesWritten: 0, quotaExceeded: false, historyId: history.historyId }; + return { + gapDetected: false, + daysTouched: [], + bytesWritten: 0, + quotaExceeded: false, + historyId: resumeFrom, + truncated: history.truncated, + }; } // Which days changed. Fetching each changed message once here — rather than @@ -1160,5 +1248,12 @@ export async function runIncrementalSync(options) { break; } } - return { gapDetected: false, daysTouched, bytesWritten, quotaExceeded, historyId: history.historyId }; + return { + gapDetected: false, + daysTouched, + bytesWritten, + quotaExceeded, + historyId: resumeFrom, + truncated: history.truncated, + }; } diff --git a/apps/mcp/test/gmailSync.test.mjs b/apps/mcp/test/gmailSync.test.mjs index 5cc2ecf52..ba6ff85c6 100644 --- a/apps/mcp/test/gmailSync.test.mjs +++ b/apps/mcp/test/gmailSync.test.mjs @@ -18,6 +18,10 @@ // sanitizeAttachmentFilename stops stripping "/" (no basename) -> 5 checks failed // resolveDayAttachments does not check `manifest.resolved` first // (always re-fetches) -> 4 checks failed +// listAllHistory reports the mailbox head after hitting maxPages +// instead of the last record it walked -> 4 checks failed +// writeDayPart puts unconditionally where the store cannot do +// a conditional write (no read-compare first) -> 2 checks failed import { GMAIL_ATTACHMENT_MAX_BYTES, @@ -375,6 +379,105 @@ export async function runGmailSyncChecks(check) { check("history.list pagination is followed and every added id collected", historyResult.messageIds.has("h1") && historyResult.messageIds.has("h2")); check("the cursor advances to the LAST page's historyId", historyResult.historyId === "1600"); + /* + A WALK THAT RAN OUT OF PAGES MUST SAY SO, AND MUST NOT HAND BACK THE HEAD. + + `history.list` returns the MAILBOX'S CURRENT `historyId` on every page, not + a per-page cursor. So a walk that stops at `maxPages` and reports that value + is reporting "you are caught up" while holding only the first N pages — + everything after them is skipped forever, silently, with no gap signalled. + A mailbox that has been quiet for weeks and then gets a first pass is + exactly where this bites. + + What it hands back instead is the last *history record's* own id, which is + a valid `startHistoryId` for the next call and covers precisely the records + this walk actually collected. That is what makes a truncated pass make + progress rather than repeating itself. + */ + /* + IDEMPOTENCE IS NOT A PROPERTY OF R2. It has to hold on the backends this + repository already says cannot do a conditional write, because a loop that + runs every few minutes against one of those is where rewriting an unchanged + day forever actually costs somebody money. + */ + const plainStore = createMemoryStore({ conditionalWrite: false }); + const plainPart = { + path: "0-inbox/email/person-at-example-invalid/2026-09-07.md", + text: "# a day\n", + }; + const firstPlainWrite = await writeDayPart(plainStore, plainPart); + const secondPlainWrite = await writeDayPart(plainStore, plainPart); + check("a first write lands on a store with no conditional write", firstPlainWrite.wrote === true); + check( + "...and re-writing the identical day there writes nothing, same as on R2", + secondPlainWrite.wrote === false, + ); + + const truncatedGmail = createFixtureGmail({ + messages: [], + history: { + pages: [ + { history: [{ id: "1100", messagesAdded: [{ message: { id: "t1" } }] }], historyId: "9999" }, + { history: [{ id: "1200", messagesAdded: [{ message: { id: "t2" } }] }], historyId: "9999" }, + { history: [{ id: "1300", messagesAdded: [{ message: { id: "t3" } }] }], historyId: "9999" }, + ], + }, + }); + const truncatedResult = await listAllHistory({ + fetchImpl: truncatedGmail.fetchImpl, + accessToken: "tok", + startHistoryId: "1000", + maxPages: 2, + }); + check("a history walk that hit its page limit reports truncation", truncatedResult.truncated === true); + check( + "...and hands back the last record it actually walked, never the mailbox head", + truncatedResult.lastRecordId === "1200" && truncatedResult.historyId === "9999", + ); + check( + "...having collected only the ids from the pages it did walk", + truncatedResult.messageIds.has("t1") && + truncatedResult.messageIds.has("t2") && + !truncatedResult.messageIds.has("t3"), + ); + + const untruncatedResult = await listAllHistory({ + fetchImpl: truncatedGmail.fetchImpl, + accessToken: "tok", + startHistoryId: "1000", + maxPages: 50, + }); + check("a walk that reached the end reports no truncation", untruncatedResult.truncated === false); + check("...and only then is the mailbox head the right place to resume", untruncatedResult.historyId === "9999"); + + const truncatedSyncGmail = createFixtureGmail({ + messages: [], + history: { + pages: [ + { history: [{ id: "1100", messagesAdded: [] }], historyId: "9999" }, + { history: [{ id: "1200", messagesAdded: [] }], historyId: "9999" }, + { history: [{ id: "1300", messagesAdded: [] }], historyId: "9999" }, + ], + }, + }); + const truncatedSync = await runIncrementalSync({ + store: createMemoryStore(), + fetchImpl: truncatedSyncGmail.fetchImpl, + accessToken: "tok", + mailboxSlug: "person-at-example-invalid", + address: "person@example.invalid", + folders: ["inbox"], + startHistoryId: "1000", + nonce: "n", + quotaBytes: 1_000_000, + maxHistoryPages: 2, + }); + check("an incremental sync tells its caller the walk was truncated", truncatedSync.truncated === true); + check( + "...and offers the record boundary as the cursor rather than the head", + truncatedSync.historyId === "1200", + ); + // -- rendering one day --------------------------------------------------------- const emptyDayParts = renderDay({ mailboxSlug: "person-at-example-invalid", address: "person@example.invalid", date: "2026-09-07", events: [], nonce: "n" }); diff --git a/apps/mobile/__tests__/communicationsPanels.test.ts b/apps/mobile/__tests__/communicationsPanels.test.ts index b6dbe582e..f893aa909 100644 --- a/apps/mobile/__tests__/communicationsPanels.test.ts +++ b/apps/mobile/__tests__/communicationsPanels.test.ts @@ -63,7 +63,18 @@ import { SettingsPane } from "../features/console/panes/SettingsPane"; import { GoogleConnectionsCard, type GoogleConnection, + type GoogleSyncSchedule, } from "../features/console/google/GoogleConnectionsCard"; + +/** A connection that is polled hourly and has actually read mail. */ +const SYNCING_HOURLY: GoogleSyncSchedule = { + intervalMinutes: 60, + everSynced: true, + cursorReady: true, + catchingUp: false, + lastAttemptAt: Date.parse("2026-09-09T09:00:00.000Z"), + nextDueAt: Date.parse("2026-09-09T10:00:00.000Z"), +}; import type { ConsoleData } from "../features/console/types"; import type { SettingsSectionKey } from "../features/console/settings/sections"; @@ -119,6 +130,7 @@ const THREE_SERVICE_ACCOUNT: GoogleConnection = { email: "someone@example.com", syncServices: { gmail: true, calendar: true, chat: true }, syncStatus: "active", + sync: SYNCING_HOURLY, gmail: { backfillDays: 90, folders: ["inbox"], @@ -225,6 +237,7 @@ describe("each panel narrows the Google card rather than repeating it", () => { workspaceId: "ws_1", disconnect: async () => null, saveDestination: async () => null, + saveSyncInterval: async () => null, }, }), ); @@ -251,6 +264,7 @@ describe("each panel narrows the Google card rather than repeating it", () => { workspaceId: "ws_1", disconnect: async () => null, saveDestination: async () => null, + saveSyncInterval: async () => null, }, connections: [THREE_SERVICE_ACCOUNT], }), diff --git a/apps/mobile/__tests__/googleConnectionsCard.test.ts b/apps/mobile/__tests__/googleConnectionsCard.test.ts index bc81a002f..a5751f76a 100644 --- a/apps/mobile/__tests__/googleConnectionsCard.test.ts +++ b/apps/mobile/__tests__/googleConnectionsCard.test.ts @@ -8,13 +8,28 @@ import { act, createElement } from "react"; import { createRoot } from "react-dom/client"; import { ThemeProvider } from "../features/design/theme"; -import { GoogleConnectionsCard } from "../features/console/google/GoogleConnectionsCard"; +import { + GoogleConnectionsCard, + type GoogleConnection, + type GoogleSyncSchedule, +} from "../features/console/google/GoogleConnectionsCard"; + +/** A connection that is polled hourly and has actually read mail. */ +const SYNCING_HOURLY: GoogleSyncSchedule = { + intervalMinutes: 60, + everSynced: true, + cursorReady: true, + catchingUp: false, + lastAttemptAt: Date.parse("2026-09-09T09:00:00.000Z"), + nextDueAt: Date.parse("2026-09-09T10:00:00.000Z"), +}; (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; const mockStartCalls: unknown[] = []; const mockNavigations: string[] = []; const mockDestinationCalls: unknown[] = []; +const mockIntervalCalls: { connectionId: string; syncIntervalMinutes: number }[] = []; jest.mock("convex/react", () => ({ useAction: () => (args: unknown) => { @@ -65,6 +80,7 @@ describe("GoogleConnectionsCard", () => { workspaceId: "ws_1", disconnect: async () => null, saveDestination: async () => null, + saveSyncInterval: async () => null, }, }), ); @@ -93,6 +109,7 @@ describe("GoogleConnectionsCard", () => { email: "seyi@supa.media", syncServices: { gmail: true, calendar: true, chat: true }, syncStatus: "connected", + sync: SYNCING_HOURLY, errorCode: "SCOPES_INCOMPLETE", lastError: "This Google account's authorization no longer covers chat.", gmail: { @@ -146,6 +163,10 @@ describe("GoogleConnectionsCard", () => { mockDestinationCalls.push({ connectionId, service, destinationPath }); return null; }, + saveSyncInterval: async (connectionId, syncIntervalMinutes) => { + mockIntervalCalls.push({ connectionId, syncIntervalMinutes }); + return null; + }, }, connections: [ { @@ -153,6 +174,7 @@ describe("GoogleConnectionsCard", () => { email: "seyi@supa.media", syncServices: { gmail: true, calendar: false, chat: false }, syncStatus: "connected", + sync: SYNCING_HOURLY, gmail: { backfillDays: 365, folders: ["inbox", "sent"], @@ -197,6 +219,7 @@ describe("GoogleConnectionsCard", () => { email: "seyi@supa.media", syncServices: { gmail: true, calendar: false, chat: false }, syncStatus: "backfilling", + sync: SYNCING_HOURLY, gmail: { backfillDays: 90, folders: ["inbox", "sent"], @@ -238,6 +261,7 @@ describe("GoogleConnectionsCard", () => { email: "seyi@supa.media", syncServices: { gmail: true, calendar: false, chat: false }, syncStatus: "active", + sync: SYNCING_HOURLY, gmail: { backfillDays: 365, folders: ["inbox"], @@ -263,3 +287,235 @@ describe("GoogleConnectionsCard", () => { screen.unmount(); }); }); + +/** + * THE SCHEDULE, WHICH IS THE PART THAT WAS MISSING. + * + * A connected mailbox that had never synced once and one syncing perfectly + * rendered the same card. These checks are what stops that returning: the + * sentence has to say which of the two this is, and the control that changes + * it has to be the owner's alone. + */ +describe("the sync schedule on a connected account", () => { + const neverSynced: GoogleSyncSchedule = { + intervalMinutes: 15, + everSynced: false, + cursorReady: false, + catchingUp: false, + }; + + function connection(sync: GoogleSyncSchedule): GoogleConnection { + return { + connectionId: "google_1", + email: "person@example.invalid", + syncServices: { gmail: true, calendar: false, chat: false }, + syncStatus: "active", + sync, + gmail: { + backfillDays: 90, + folders: ["inbox", "sent"], + destinationFolder: "0-inbox/email/person-at-example-invalid", + destinationPath: "0-inbox/email/person-at-example-invalid/YYYY-MM-DD.md", + historyCursorReady: true, + }, + }; + } + + test("a connection that has never synced says so, rather than looking healthy", () => { + const screen = render( + createElement(GoogleConnectionsCard, { + service: "gmail", + connections: [connection(neverSynced)], + }), + ); + const text = screen.container.textContent ?? ""; + expect(text).toContain("Every 15 min"); + expect(text).toContain("never synced yet"); + expect(text).toContain("due now"); + screen.unmount(); + }); + + test("...and one that has synced does not", () => { + const screen = render( + createElement(GoogleConnectionsCard, { + service: "gmail", + connections: [connection(SYNCING_HOURLY)], + }), + ); + const text = screen.container.textContent ?? ""; + expect(text).toContain("Every 1 hour"); + expect(text).not.toContain("never synced"); + screen.unmount(); + }); + + test("a pass that has only ever failed is not reported as never having been tried", () => { + const screen = render( + createElement(GoogleConnectionsCard, { + service: "gmail", + connections: [ + connection({ + intervalMinutes: 15, + everSynced: false, + cursorReady: false, + catchingUp: false, + lastAttemptAt: Date.parse("2026-09-09T09:00:00.000Z"), + nextDueAt: Date.parse("2026-09-09T09:15:00.000Z"), + lastFailureAt: Date.parse("2026-09-09T09:00:00.000Z"), + lastFailureCode: "GOOGLE_ACCESS_REFUSED", + lastFailure: "Google refused access to this mailbox.", + }), + ], + }), + ); + const text = screen.container.textContent ?? ""; + expect(text).toContain("has not synced successfully yet"); + expect(text).toContain("Google refused access to this mailbox."); + screen.unmount(); + }); + + test("a Calendar panel is not shown a schedule that only advances mail", () => { + const account = connection(neverSynced); + const screen = render( + createElement(GoogleConnectionsCard, { + service: "calendar", + connections: [ + { + ...account, + syncServices: { gmail: true, calendar: true, chat: false }, + calendar: { + destinationFolder: "0-inbox/calendar", + destinationPath: "0-inbox/calendar/YYYY-MM-DD.md", + syncCursorReady: false, + }, + }, + ], + }), + ); + const text = screen.container.textContent ?? ""; + expect(text).not.toContain("Sync schedule"); + expect(text).not.toContain("Every 15 min"); + // Calendar's own honest sentence is still there. + expect(text).toContain("Connected; upcoming event sync setup is pending"); + screen.unmount(); + }); + + test("a baselined mailbox says it is watching, not that it has synced", () => { + const screen = render( + createElement(GoogleConnectionsCard, { + service: "gmail", + connections: [ + connection({ + intervalMinutes: 15, + everSynced: false, + cursorReady: true, + catchingUp: false, + lastAttemptAt: Date.parse("2026-09-09T09:00:00.000Z"), + nextDueAt: Date.parse("2026-09-09T09:15:00.000Z"), + }), + ], + }), + ); + const text = screen.container.textContent ?? ""; + expect(text).toContain("watching for new mail; none read yet"); + expect(text).not.toContain("never synced yet"); + screen.unmount(); + }); + + test("a mailbox draining a backlog says so instead of naming a next due time", () => { + const screen = render( + createElement(GoogleConnectionsCard, { + service: "gmail", + connections: [ + connection({ + intervalMinutes: 60, + everSynced: true, + cursorReady: true, + catchingUp: true, + lastAttemptAt: Date.parse("2026-09-09T09:00:00.000Z"), + nextDueAt: Date.parse("2026-09-09T09:00:00.000Z"), + }), + ], + }), + ); + const text = screen.container.textContent ?? ""; + expect(text).toContain("catching up on older mail"); + screen.unmount(); + }); + + test("the picker is the owner's alone — absent for anybody else, not disabled", () => { + const withoutActions = render( + createElement(GoogleConnectionsCard, { + service: "gmail", + connections: [connection(neverSynced)], + }), + ); + expect( + withoutActions.container.querySelector('[data-testid="google-sync-interval-5-google_1"]'), + ).toBeNull(); + // The status itself is still shown: somebody who cannot change the + // schedule can still need to know the last pass failed. + expect(withoutActions.container.textContent).toContain("Every 15 min"); + withoutActions.unmount(); + + const asOwner = render( + createElement(GoogleConnectionsCard, { + service: "gmail", + connections: [connection(neverSynced)], + actions: { + workspaceId: "ws_1", + disconnect: async () => null, + saveDestination: async () => null, + saveSyncInterval: async () => null, + }, + }), + ); + expect( + asOwner.container.querySelector('[data-testid="google-sync-interval-5-google_1"]'), + ).not.toBeNull(); + asOwner.unmount(); + }); + + test("choosing an interval sends exactly that many minutes", async () => { + mockIntervalCalls.length = 0; + const screen = render( + createElement(GoogleConnectionsCard, { + service: "gmail", + connections: [connection(neverSynced)], + actions: { + workspaceId: "ws_1", + disconnect: async () => null, + saveDestination: async () => null, + saveSyncInterval: async (connectionId, syncIntervalMinutes) => { + mockIntervalCalls.push({ connectionId, syncIntervalMinutes }); + return null; + }, + }, + }), + ); + await screen.click("google-sync-interval-5-google_1"); + expect(mockIntervalCalls).toEqual([{ connectionId: "google_1", syncIntervalMinutes: 5 }]); + screen.unmount(); + }); + + test("a schedule the server refuses is shown, not swallowed", async () => { + const screen = render( + createElement(GoogleConnectionsCard, { + service: "gmail", + connections: [connection(neverSynced)], + actions: { + workspaceId: "ws_1", + disconnect: async () => null, + saveDestination: async () => null, + saveSyncInterval: async () => { + throw new Error("Sync can run at most every 5 minutes."); + }, + }, + }), + ); + await screen.click("google-sync-interval-1440-google_1"); + const text = screen.container.textContent ?? ""; + expect(text).toContain("Schedule was not saved"); + expect(text).toContain("Sync can run at most every 5 minutes."); + screen.unmount(); + }); +}); diff --git a/apps/mobile/features/console/advanced/advanced.ts b/apps/mobile/features/console/advanced/advanced.ts index c2f92df2b..6ffdbb36f 100644 --- a/apps/mobile/features/console/advanced/advanced.ts +++ b/apps/mobile/features/console/advanced/advanced.ts @@ -135,6 +135,8 @@ const ACTION_LABELS: Readonly> = { "mail.disconnected": "Disconnected mail", "mail.rekeyed": "Rotated a mail credential", "google_sync_destination_updated": "Changed a Google sync destination", + "google_sync_interval_updated": "Changed how often Google syncs", + "google_sync_wrote": "Synced mail into this context", "encryption.export": "Exported this context's encryption keys", "encryption.rekeyed": "Rotated an encryption key", }; diff --git a/apps/mobile/features/console/google/GoogleConnectionsCard.tsx b/apps/mobile/features/console/google/GoogleConnectionsCard.tsx index 9e9ffbca2..61a2ce9b6 100644 --- a/apps/mobile/features/console/google/GoogleConnectionsCard.tsx +++ b/apps/mobile/features/console/google/GoogleConnectionsCard.tsx @@ -10,11 +10,54 @@ import { useArming } from "../useArming"; import { GOOGLE_REDIRECT_ORIGINS, type GoogleSyncServices } from "./google"; import { useGoogleStart } from "./useGoogleStart"; +/** + * The floor, and the choices offered for it. + * + * The floor is the server's (`functions/googleSync.ts`, + * `MIN_SYNC_INTERVAL_MINUTES`) and is restated here only to build the picker — + * a value typed past it is refused by the mutation, not by this list, which is + * why the refusal is shown rather than prevented. + */ +export const SYNC_INTERVAL_CHOICES = [5, 15, 30, 60, 240, 1440] as const; + +export interface GoogleSyncSchedule { + intervalMinutes: number; + /** Has a pass ever actually read mail from this account? */ + everSynced: boolean; + /** + * A cursor exists, so the next pass will read whatever arrives. + * + * Separate from `everSynced` because forward-only makes them genuinely + * different: the pass that establishes a cursor reads nothing, by design, + * and a card that treated it as a sync would show a mailbox as current + * before a single message had been read — the same confusion this block + * exists to end, one state later. + */ + cursorReady: boolean; + /** The last pass ran out of history pages and there is more to drain. */ + catchingUp: boolean; + /** When a pass last finished, successfully or not. */ + lastAttemptAt?: number; + nextDueAt?: number; + lastFailureAt?: number; + lastFailureCode?: string; + lastFailure?: string; +} + export interface GoogleConnection { connectionId: string; email: string; syncServices: GoogleSyncServices; syncStatus: string; + /** + * How often this account is polled, and how the last poll went. + * + * Required rather than optional on purpose: every site that builds one of + * these has to say whether this connection has ever synced, because "it + * looks connected" and "it is actually syncing" being the same screen is the + * defect this card is being changed to fix. + */ + sync: GoogleSyncSchedule; lastSyncStartedAt?: number; lastSyncCompletedAt?: number; errorCode?: string; @@ -67,6 +110,12 @@ export interface GoogleActions { service: "gmail" | "calendar" | "chat", destinationPath: string, ) => Promise; + /** + * How often this account is polled. Owner-only like every other action on + * this object — the whole object is absent for anybody else, which is how + * the control ends up *absent* rather than disabled. + */ + saveSyncInterval: (connectionId: string, syncIntervalMinutes: number) => Promise; } /** @@ -401,6 +450,21 @@ function ConnectedGoogleRow({ /> ) : null} + {/* + The schedule is the *account's* — one grant, one pass — but only + Gmail is advanced by that pass today, so it is drawn only where Gmail + is in view. A Calendar or Chat panel showing "every 15 minutes, next + due at 10:15" would be a promise this loop does not yet keep for + those two; their own status lines already say their sync is pending. + When they join the loop, this condition is what goes. + */} + {connection.syncServices.gmail && showBlock("gmail") ? ( + + ) : null} {showAccountError ? ( (null); + const [saveError, setSaveError] = useState(null); + + return ( + + Sync schedule + + {describeSchedule(sync)} + + {saveSyncInterval ? ( + + {SYNC_INTERVAL_CHOICES.map((minutes) => ( +