From f5fef340889a4f66e2eb02088955e829dc78d157 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 04:08:44 +0000 Subject: [PATCH 01/14] Advance a connected Google account's cursor on a schedule MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A person connects Gmail, the grant and a historyId baseline are stored, and nothing has ever advanced that cursor: no cron entry, no webhook, and the gateway's scheduled() handler has no trigger configured. The Gmail pipeline that would have done the work — apps/mcp/src/communications/gmailSync.js — was imported by nothing at all after #388 removed the historical backfill that used to import it. This is the trigger that was missing, not a second pipeline: - a five-minute cron sweeping connections that are actually due (now >= lastSyncAt + interval), claiming each one so a pass still running is never overtaken, and holding no decision of its own; - a per-connection syncIntervalMinutes, owner-only, floored at five minutes server-side and defaulting to fifteen; - one forward pass per account inside the existing credential barrier, which re-asks every gate before it opens a credential and advances historyId only over mail it actually wrote. Forward-only per #388: an expired cursor is re-baselined and the gap is recorded, never backfilled. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01W42G6GPAm2Nh3nCxp2ir9D --- apps/convex/_generated/api.d.ts | 2 + apps/convex/crons.ts | 35 ++ apps/convex/functions/files.ts | 420 +++++++++++++++++++ apps/convex/functions/googleConnect.ts | 9 +- apps/convex/functions/googleSync.ts | 550 +++++++++++++++++++++++++ apps/convex/schema.ts | 67 ++- 6 files changed, 1081 insertions(+), 2 deletions(-) create mode 100644 apps/convex/functions/googleSync.ts diff --git a/apps/convex/_generated/api.d.ts b/apps/convex/_generated/api.d.ts index 9ba2598af..c6c9a459e 100644 --- a/apps/convex/_generated/api.d.ts +++ b/apps/convex/_generated/api.d.ts @@ -66,6 +66,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"; @@ -141,6 +142,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..08a8a1326 100644 --- a/apps/convex/crons.ts +++ b/apps/convex/crons.ts @@ -146,4 +146,39 @@ 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*. + * + * 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 50e63b399..030c5727a 100644 --- a/apps/convex/functions/files.ts +++ b/apps/convex/functions/files.ts @@ -83,6 +83,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, @@ -441,6 +457,23 @@ 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(), + errorCode: v.optional(v.string()), +}); + const operationResultValidator = v.union( listingValidator, fileValidator, @@ -457,6 +490,7 @@ const operationResultValidator = v.union( indexMaintainedValidator, indexProjectedValidator, googleSyncRunValidator, + googleForwardSyncValidator, ); const operationValidator = v.union( @@ -507,6 +541,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(), @@ -614,6 +655,16 @@ type OperationResult = bytesWritten: number; continue: boolean; } + | { + kind: "googleForwardSync"; + connectionId: Id<"googleConnections">; + status: "synced" | "skipped" | "failed"; + daysTouched: number; + bytesWritten: number; + cursorAdvanced: boolean; + gapDetected: boolean; + errorCode?: string; + } | { kind: "listing"; path: string; @@ -878,12 +929,47 @@ 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, + { connectionId: args.operation.connectionId }, + ); + if (forwardSyncJob === null || forwardSyncJob.kind === "skip") { + return await releaseForwardSync( + ctx, + args.operation.connectionId, + forwardSyncJob === null ? undefined : 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: @@ -891,6 +977,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: @@ -932,6 +1026,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: @@ -939,6 +1041,10 @@ export const runFileOperation = internalAction({ }); } + if (args.operation.kind === "googleForwardSync" && forwardSyncJob?.kind === "run") { + return await runGoogleForwardSync(ctx, store, forwardSyncJob, Date.now()); + } + const result = await executeOperation( store, args.scope, @@ -1047,6 +1153,320 @@ 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, + 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, + 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, + now: number, +): Promise { + 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.", + ); + } + + try { + 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, + }); + return { + kind: "googleForwardSync", + connectionId: job.connectionId, + status: "synced", + daysTouched: 0, + bytesWritten: 0, + cursorAdvanced: true, + gapDetected: 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}`, + now: new Date(now).toISOString(), + 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, + }); + return { + kind: "googleForwardSync", + connectionId: job.connectionId, + status: "synced", + daysTouched: 0, + bytesWritten: 0, + cursorAdvanced: true, + gapDetected: true, + }; + } + + 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, + errorCode: "MAIL_QUOTA_EXCEEDED", + }; + } + + await ctx.runMutation(internal.functions.googleSync.recordGoogleForwardSyncPass, { + connectionId: job.connectionId, + status: "synced", + historyId: result.historyId, + daysTouched: result.daysTouched.length, + bytesWritten: result.bytesWritten, + }); + return { + kind: "googleForwardSync", + connectionId: job.connectionId, + status: "synced", + daysTouched: result.daysTouched.length, + bytesWritten: result.bytesWritten, + cursorAdvanced: result.historyId !== undefined, + gapDetected: false, + }; + } 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..39720f296 100644 --- a/apps/convex/functions/googleConnect.ts +++ b/apps/convex/functions/googleConnect.ts @@ -217,7 +217,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 { diff --git a/apps/convex/functions/googleSync.ts b/apps/convex/functions/googleSync.ts new file mode 100644 index 000000000..9be3a5936 --- /dev/null +++ b/apps/convex/functions/googleSync.ts @@ -0,0 +1,550 @@ +/** + * 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 type { Doc } from "../_generated/dataModel"; +import { recordAudit } from "./lib/audit"; +import { + defaultGoogleDestinationFolder, + mailConnectEnabled, + requireActor, +} from "./googleConnect"; + +/** + * The lowest interval this deployment will accept, in minutes — **refused + * server-side**, not merely absent from the 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; + +/** Connections one sweep may start, bounded like every other sweep here. */ +const SWEEP_BATCH = 50; + +/** + * Which products this engine 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 this module's entry in `docs/decisions/communications.md` for what each + * of them still needs before that is true. + */ +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 this engine 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 }, + now: number, +): boolean { + if (connection.lastSyncAt === undefined) return true; + return now >= connection.lastSyncAt + syncIntervalMinutesOf(connection) * 60_000; +} + +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 this + // `sweepStalledBackfills` wears over `optedIn`. + 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: { connectionId: v.id("googleConnections") }, + returns: forwardSyncJobValidator, + handler: async (ctx, args) => { + const connection = await ctx.db.get(args.connectionId); + if (connection === null) 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(connection.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()), + 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") { + await ctx.db.patch(args.connectionId, { + syncStartedAt: undefined, + lastSyncAt: now, + nextSyncAt: now + Math.max(intervalMs, SYNC_FAILURE_BACKOFF_MS), + health: "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, + 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; + await ctx.db.patch(args.connectionId, { + gmail, + syncStartedAt: undefined, + lastSyncAt: now, + nextSyncAt: now + intervalMs, + 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, + }); + return { accepted: true }; + }, +}); + +/** The row shape the console reads, kept in one place for `listGoogleConnections`. */ +export function syncStatusOf(connection: Doc<"googleConnections">): { + intervalMinutes: number; + everSynced: 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, + 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 6bc25092e..0e48afd31 100644 --- a/apps/convex/schema.ts +++ b/apps/convex/schema.ts @@ -1034,6 +1034,58 @@ 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. + * + * These five fields 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()), + /** + * 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()), + 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()), /** * Set by disconnect. The row is kept — never deleted outright — so a * disconnected connection's sync job can be told apart from one that @@ -1049,7 +1101,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. From 27e36225d129755b8802db994fc3acbd7469a85c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 04:13:08 +0000 Subject: [PATCH 02/14] Prove the sync loop's guards, and stop it rewriting unchanged days MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Forty-three checks over the sweep's due selection, the five-minute floor, the owner gate (attacker and victim in one database), the not-overtaking guard, and the cursor — including eight that drive the real runFileOperation against a fixture Gmail and an in-memory S3, so the pass under test is the pass that ships. The idempotence check found a defect the historical backfill also had: passing a wall-clock `now` into runIncrementalSync travels straight through syncOneDay to renderDay, overriding the `updated` default that is keyed to the newest message's own timestamp. A one-shot import survives that; a pass that runs every few minutes would rewrite every touched day forever. The loop passes no `now`, and "re-running the same pass writes no new bytes" is the check that holds it. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01W42G6GPAm2Nh3nCxp2ir9D --- apps/convex/__tests__/googleSyncLoop.test.ts | 971 +++++++++++++++++++ apps/convex/functions/files.ts | 17 +- 2 files changed, 985 insertions(+), 3 deletions(-) create mode 100644 apps/convex/__tests__/googleSyncLoop.test.ts diff --git a/apps/convex/__tests__/googleSyncLoop.test.ts b/apps/convex/__tests__/googleSyncLoop.test.ts new file mode 100644 index 000000000..bcf08f2df --- /dev/null +++ b/apps/convex/__tests__/googleSyncLoop.test.ts @@ -0,0 +1,971 @@ +/** + * 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_INTERVAL_MINUTES, + MIN_SYNC_INTERVAL_MINUTES, + SYNC_FAILURE_BACKOFF_MS, + SYNC_STALL_MS, +} from "../functions/googleSync"; + +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", + 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. + expect(row.nextSyncAt).toBe(row.lastSyncAt! + 30 * 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); + expect(row.nextSyncAt).toBe(row.lastSyncAt! + SYNC_FAILURE_BACKOFF_MS); + }); + + 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("the pass re-asks every gate before it opens a credential", () => { + beforeEach(() => enableMailSync()); + + test("a disconnected account is skipped", async () => { + const { t, connectionId } = await scenario(); + await patchConnection(t, connectionId, { disconnectedAt: Date.now() }); + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { connectionId }), + ).toEqual({ kind: "skip", reason: "GOOGLE_DISCONNECTED" }); + }); + + test("a deployment that may not read mail is skipped even mid-flight", async () => { + const { t, connectionId } = await scenario(); + vi.stubEnv("MAIL_CONNECT_ENABLED", ""); + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { 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, connectionId } = await scenario(); + await patchConnection(t, connectionId, { health: "reconnect_required" }); + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { connectionId }), + ).toEqual({ kind: "skip", reason: "GOOGLE_RECONNECT_REQUIRED" }); + }); + + test("a product turned off since the sweep looked is skipped", async () => { + const { t, connectionId } = await scenario(); + await patchConnection(t, connectionId, { products: [] }); + expect( + await t.query(internal.functions.googleSync.googleForwardSyncJob, { 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, { connectionId }), + ).toEqual({ kind: "skip", reason: "NOT_PERSONAL_CONTEXT" }); + }); + + test("a live connection is handed its settings and cursor, and no secret", async () => { + const { t, connectionId } = await scenario(); + await patchConnection(t, connectionId, { + gmail: { ...(await readConnection(t, connectionId)).gmail!, historyId: "1000" }, + }); + const job = await t.query(internal.functions.googleSync.googleForwardSyncJob, { + 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"); + }); +}); + +/* -------------------------------------------------------------------------- */ +/* 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 }; + 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); + 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 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("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("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("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 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/functions/files.ts b/apps/convex/functions/files.ts index 030c5727a..7fe8e1567 100644 --- a/apps/convex/functions/files.ts +++ b/apps/convex/functions/files.ts @@ -1042,7 +1042,7 @@ export const runFileOperation = internalAction({ } if (args.operation.kind === "googleForwardSync" && forwardSyncJob?.kind === "run") { - return await runGoogleForwardSync(ctx, store, forwardSyncJob, Date.now()); + return await runGoogleForwardSync(ctx, store, forwardSyncJob); } const result = await executeOperation( @@ -1326,7 +1326,6 @@ async function runGoogleForwardSync( ctx: ActionCtx, store: FileStore, job: Extract, - now: number, ): Promise { const minted = await ctx.runAction(internal.functions.googleConnect.mintGoogleAccessToken, { connectionId: job.connectionId, @@ -1378,7 +1377,19 @@ async function runGoogleForwardSync( // 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}`, - now: new Date(now).toISOString(), + /* + * 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, From fdcf2e8a1b9e97b16ec47f32287ce2ac974f3d9a Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 04:19:00 +0000 Subject: [PATCH 03/14] Show whether a connected account is actually syncing, and let its owner say how often MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A connected mailbox that had never synced once and one syncing perfectly rendered the same card, which is exactly the state every connection in this product was in. The connection listing now carries the schedule — how often it is polled, whether it has ever read mail, when it is next due, and the last failure, kept after a later pass succeeded — and the card says which of the two states this is in one sentence. The interval picker is owner-only by being drawn only where GoogleActions is, which is absent rather than disabled for anybody else. It starts at the floor and does not enforce it: a value below five minutes is refused by the mutation and the refusal is rendered, so the check that matters is the server's. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01W42G6GPAm2Nh3nCxp2ir9D --- apps/convex/__tests__/googleSyncLoop.test.ts | 94 +++++++++ apps/convex/functions/googleConnect.ts | 25 +++ .../__tests__/communicationsPanels.test.ts | 12 ++ .../__tests__/googleConnectionsCard.test.ts | 180 +++++++++++++++++- .../console/google/GoogleConnectionsCard.tsx | 153 +++++++++++++++ .../features/console/useLiveConsoleData.ts | 20 ++ 6 files changed, 483 insertions(+), 1 deletion(-) diff --git a/apps/convex/__tests__/googleSyncLoop.test.ts b/apps/convex/__tests__/googleSyncLoop.test.ts index bcf08f2df..355be6244 100644 --- a/apps/convex/__tests__/googleSyncLoop.test.ts +++ b/apps/convex/__tests__/googleSyncLoop.test.ts @@ -630,6 +630,100 @@ describe("the pass re-asks every gate before it opens a credential", () => { }); }); +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 */ /* -------------------------------------------------------------------------- */ diff --git a/apps/convex/functions/googleConnect.ts b/apps/convex/functions/googleConnect.ts index 39720f296..d736da417 100644 --- a/apps/convex/functions/googleConnect.ts +++ b/apps/convex/functions/googleConnect.ts @@ -64,6 +64,11 @@ 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 lives in `googleSync.ts` — the loop that +// advances it — and the console reads both halves off one row. Importing the +// view rather than re-deriving it here keeps "when is this next due" from +// having two implementations that can disagree. +import { syncStatusOf } from "./googleSync"; import { createPkcePair, exchangeGoogleCode, @@ -775,6 +780,25 @@ 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(), + 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( @@ -877,6 +901,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/mobile/__tests__/communicationsPanels.test.ts b/apps/mobile/__tests__/communicationsPanels.test.ts index b6dbe582e..211dae921 100644 --- a/apps/mobile/__tests__/communicationsPanels.test.ts +++ b/apps/mobile/__tests__/communicationsPanels.test.ts @@ -63,7 +63,16 @@ 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, + 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 +128,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 +235,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 +262,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..275c427a8 100644 --- a/apps/mobile/__tests__/googleConnectionsCard.test.ts +++ b/apps/mobile/__tests__/googleConnectionsCard.test.ts @@ -8,13 +8,26 @@ 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, + 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 +78,7 @@ describe("GoogleConnectionsCard", () => { workspaceId: "ws_1", disconnect: async () => null, saveDestination: async () => null, + saveSyncInterval: async () => null, }, }), ); @@ -93,6 +107,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 +161,10 @@ describe("GoogleConnectionsCard", () => { mockDestinationCalls.push({ connectionId, service, destinationPath }); return null; }, + saveSyncInterval: async (connectionId, syncIntervalMinutes) => { + mockIntervalCalls.push({ connectionId, syncIntervalMinutes }); + return null; + }, }, connections: [ { @@ -153,6 +172,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 +217,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 +259,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 +285,159 @@ 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 }; + + 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, + 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("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/google/GoogleConnectionsCard.tsx b/apps/mobile/features/console/google/GoogleConnectionsCard.tsx index 9e9ffbca2..302fd135d 100644 --- a/apps/mobile/features/console/google/GoogleConnectionsCard.tsx +++ b/apps/mobile/features/console/google/GoogleConnectionsCard.tsx @@ -10,11 +10,42 @@ 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 from this account? */ + everSynced: 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 +98,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 +438,11 @@ function ConnectedGoogleRow({ /> ) : null} + {showAccountError ? ( (null); + const [saveError, setSaveError] = useState(null); + + return ( + + Sync schedule + + {describeSchedule(sync)} + + {saveSyncInterval ? ( + + {SYNC_INTERVAL_CHOICES.map((minutes) => ( +