diff --git a/infra/lambdas/google-content-sync/changes.ts b/infra/lambdas/google-content-sync/changes.ts new file mode 100644 index 000000000..3ffcbdb02 --- /dev/null +++ b/infra/lambdas/google-content-sync/changes.ts @@ -0,0 +1,57 @@ +/** + * Classification helpers for entries returned by the Google Drive Changes + * feed (`GET /drive/v3/changes`). + * + * The feed is not exclusively file-scoped. Google emits `changeType: "drive"` + * entries for Shared Drive-level events — rename, membership change, + * restriction change — and those entries carry NO `fileId`. There is no file + * to import, resolve, or mark missing for them. + * + * Every downstream step of the reconcile loop is keyed on a file id, so a + * `fileId`-less entry must be skipped before that work begins. `changeType` is + * matched as a plain string rather than an enum so a future value Google adds + * cannot turn the queue into poison messages again. + */ + +/** Shared Drive-scoped change entries never carry a `fileId`. */ +export const DRIVE_SCOPED_CHANGE_TYPE = "drive"; + +/** The minimum shape needed to decide whether a change entry is actionable. */ +export interface DriveChangeIdentity { + changeType?: string | undefined; + fileId?: string | undefined; + removed?: boolean | undefined; +} + +/** + * True when the entry names a specific Drive file, i.e. it is safe to hand to + * the file-scoped import/removal path. Drive-scoped entries and entries with a + * missing or empty `fileId` return false and must be skipped. + */ +export function isFileScopedDriveChange( + change: T, +): change is T & { fileId: string } { + if (change.changeType === DRIVE_SCOPED_CHANGE_TYPE) return false; + return typeof change.fileId === "string" && change.fileId.length > 0; +} + +/** + * True when a non-file-scoped entry reports that the Shared Drive itself went + * away — deleted, or this account lost access to it. + * + * These entries must NOT be skipped like the other drive-scoped noise. Sources + * already imported from that drive are now unreachable, and a connector + * reading the global changes feed gets no other signal: its subsequent + * `listChanges` and watch calls keep succeeding, so the stale content would + * stay active indefinitely. The caller answers this by requesting a selection + * snapshot, which retires whatever is no longer reachable. + * + * Deliberately not keyed on `changeType === "drive"`: the legacy "teamDrive" + * value and any future scope Google introduces must take this path too. What + * identifies the case is "removed, and not about a specific file" — which is + * exactly the state the caller has already established. + */ +export function isDriveRemovalChange(change: DriveChangeIdentity): boolean { + if (isFileScopedDriveChange(change)) return false; + return change.removed === true; +} diff --git a/infra/lambdas/google-content-sync/index.ts b/infra/lambdas/google-content-sync/index.ts index b45cfff5d..123d656ce 100644 --- a/infra/lambdas/google-content-sync/index.ts +++ b/infra/lambdas/google-content-sync/index.ts @@ -76,6 +76,7 @@ import { } from "../../../lib/repositories/google-drive/formats"; import { refreshGoogleAccessToken } from "../../../lib/repositories/google-drive/oauth"; import { getGoogleContentWifAccessToken } from "../../../lib/repositories/google-drive/wif"; +import { isDriveRemovalChange, isFileScopedDriveChange } from "./changes"; import { assertGoogleSourceMetadataSize, assertGoogleSourceResponseSize, @@ -1332,12 +1333,62 @@ type GoogleDriveChange = Awaited< ReturnType >["values"][number]; +/** + * Records the outcome of a failed file-scoped change. Connector lifecycle + * errors are fatal to the whole run and are rethrown; a Drive 403/404 means the + * file was deleted or access was lost, which is the markSourceMissing path. + * Anything else is rethrown so the caller can fail the record. + */ +async function recordDriveChangeFailure( + context: ConnectorContext, + fileId: string, + error: unknown, + counters: SyncCounters, +): Promise { + if ( + error instanceof ConnectorRevokedError || + error instanceof ConnectorPausedError || + error instanceof ConnectorSelectionChangedError + ) { + throw error; + } + if (!isGoogleDriveMissingError(error)) throw error; + if (await markSourceMissing(context, fileId)) { + counters.missing += 1; + } else { + counters.failed += 1; + } +} + async function processGoogleDriveChange( context: ConnectorContext, client: GoogleDriveClient, change: GoogleDriveChange, counters: SyncCounters, ): Promise { + // Shared Drive-scoped entries (rename, membership, restriction changes) have + // no fileId, so nothing on the file path below can act on them. + if (!isFileScopedDriveChange(change)) { + // A removal, though, is not noise: the Shared Drive was deleted or this + // account lost access, so sources imported from it are now unreachable. + // Requesting a selection snapshot lets markUnseenSourcesMissing retire + // them; a hard access loss still surfaces through the existing 403/404 + // path. Anything else is metadata noise and is skipped without side + // effects so the cursor keeps advancing. + const driveRemoved = isDriveRemovalChange(change); + log.info( + driveRemoved + ? "Shared Drive removed from the change feed" + : "Skipped non-file Google Drive change entry", + { + connectorId: context.connector.id, + changeType: change.changeType ?? null, + driveId: change.driveId ?? null, + time: change.time ?? null, + }, + ); + return driveRemoved; + } if (change.removed || !change.file || change.file.trashed) { if (await markSourceMissing(context, change.fileId)) { counters.missing += 1; @@ -1363,21 +1414,7 @@ async function processGoogleDriveChange( ); return false; } catch (error) { - if ( - error instanceof ConnectorRevokedError || - error instanceof ConnectorPausedError || - error instanceof ConnectorSelectionChangedError - ) { - throw error; - } - if (!isGoogleDriveMissingError(error)) throw error; - if ( - await markSourceMissing(context, change.fileId) - ) { - counters.missing += 1; - } else { - counters.failed += 1; - } + await recordDriveChangeFailure(context, change.fileId, error, counters); return false; } } diff --git a/lib/repositories/google-drive/drive-client.ts b/lib/repositories/google-drive/drive-client.ts index b6ad170b2..473a1a781 100644 --- a/lib/repositories/google-drive/drive-client.ts +++ b/lib/repositories/google-drive/drive-client.ts @@ -56,8 +56,19 @@ const filesListSchema = z.object({ files: z.array(googleDriveFileSchema).default([]), }); +/** + * Google's Changes feed is not exclusively file-scoped. Entries with + * `changeType: "drive"` describe the Shared Drive itself (rename, membership, + * restriction changes) and carry NO `fileId`. Requiring `fileId` here made + * every such entry a poison message: `listChanges` threw before the cursor + * advanced, so the Lambda retried the same page forever. + * + * `changeType` is deliberately a plain string rather than an enum — a new + * value Google adds later must not re-poison the queue. + */ const driveChangeSchema = z.object({ - fileId: z.string(), + changeType: z.string().optional(), + fileId: z.string().optional(), removed: z.boolean().optional().default(false), time: z.string().datetime({ offset: true }).optional(), driveId: z.string().optional(), @@ -267,7 +278,7 @@ export class GoogleDriveClient { url.searchParams.set("includeItemsFromAllDrives", "true"); url.searchParams.set( "fields", - `nextPageToken,newStartPageToken,changes(fileId,removed,time,driveId,file(${FILE_FIELDS}))`, + `nextPageToken,newStartPageToken,changes(changeType,fileId,removed,time,driveId,file(${FILE_FIELDS}))`, ); if (driveId) { url.searchParams.set("driveId", driveId); diff --git a/tests/unit/google-content-sync-change-scope.test.ts b/tests/unit/google-content-sync-change-scope.test.ts new file mode 100644 index 000000000..d8bb58db9 --- /dev/null +++ b/tests/unit/google-content-sync-change-scope.test.ts @@ -0,0 +1,168 @@ +/** @jest-environment node */ + +import fs from "node:fs"; +import path from "node:path"; +import { + DRIVE_SCOPED_CHANGE_TYPE, + isDriveRemovalChange, + isFileScopedDriveChange, +} from "../../infra/lambdas/google-content-sync/changes"; +import { stripComments } from "../helpers/strip-ts-comments"; + +describe("Google Drive change scope classification", () => { + test("skips Shared Drive-scoped entries even when a fileId is present", () => { + expect( + isFileScopedDriveChange({ changeType: DRIVE_SCOPED_CHANGE_TYPE }), + ).toBe(false); + expect( + isFileScopedDriveChange({ + changeType: DRIVE_SCOPED_CHANGE_TYPE, + fileId: "file-1", + }), + ).toBe(false); + }); + + test("skips entries with a missing or empty fileId", () => { + expect(isFileScopedDriveChange({})).toBe(false); + expect(isFileScopedDriveChange({ changeType: "file" })).toBe(false); + expect(isFileScopedDriveChange({ fileId: "" })).toBe(false); + expect(isFileScopedDriveChange({ fileId: undefined })).toBe(false); + }); + + test("processes file-scoped entries, including ones with no changeType", () => { + expect(isFileScopedDriveChange({ changeType: "file", fileId: "f1" })).toBe( + true, + ); + expect(isFileScopedDriveChange({ fileId: "f1" })).toBe(true); + }); + + test("treats unknown change types as file-scoped when a fileId is present", () => { + // A future changeType value must not become a poison message: if Google + // gave us a fileId, the file path can still act on it. + expect( + isFileScopedDriveChange({ changeType: "someFutureScope", fileId: "f1" }), + ).toBe(true); + expect(isFileScopedDriveChange({ changeType: "someFutureScope" })).toBe( + false, + ); + }); + + test("narrows fileId to a string for the file-scoped branch", () => { + const change: { changeType?: string; fileId?: string } = { + changeType: "file", + fileId: "file-1", + }; + if (isFileScopedDriveChange(change)) { + const fileId: string = change.fileId; + expect(fileId).toBe("file-1"); + } else { + throw new Error("expected the entry to be file-scoped"); + } + }); +}); + +describe("Shared Drive removal detection", () => { + test("flags a drive-scoped removal so the caller rebuilds the selection", () => { + // The Shared Drive was deleted or access was lost. Sources imported from + // it are unreachable, and a connector reading the global changes feed gets + // no other signal — its later listChanges calls keep succeeding. + expect( + isDriveRemovalChange({ + changeType: DRIVE_SCOPED_CHANGE_TYPE, + removed: true, + }), + ).toBe(true); + }); + + test("covers legacy and future drive-like scopes, not just \"drive\"", () => { + expect( + isDriveRemovalChange({ changeType: "teamDrive", removed: true }), + ).toBe(true); + expect( + isDriveRemovalChange({ changeType: "someFutureScope", removed: true }), + ).toBe(true); + expect(isDriveRemovalChange({ removed: true })).toBe(true); + }); + + test("ignores drive-scoped metadata noise", () => { + // Rename, membership and restriction changes carry removed: false. + expect( + isDriveRemovalChange({ + changeType: DRIVE_SCOPED_CHANGE_TYPE, + removed: false, + }), + ).toBe(false); + expect( + isDriveRemovalChange({ changeType: DRIVE_SCOPED_CHANGE_TYPE }), + ).toBe(false); + }); + + test("never claims a file-scoped removal — that is the file path's job", () => { + expect( + isDriveRemovalChange({ + changeType: "file", + fileId: "file-1", + removed: true, + }), + ).toBe(false); + }); + + test("treats a drive-scoped entry that also carries a fileId as drive-scoped", () => { + expect( + isDriveRemovalChange({ + changeType: DRIVE_SCOPED_CHANGE_TYPE, + fileId: "file-1", + removed: true, + }), + ).toBe(true); + }); +}); + +describe("processGoogleDriveChange guard wiring", () => { + const source = stripComments( + fs.readFileSync( + path.join(process.cwd(), "infra/lambdas/google-content-sync/index.ts"), + "utf8", + ), + ); + + test("guards the change handler before any file-scoped work", () => { + const body = source.slice( + source.indexOf("async function processGoogleDriveChange("), + ); + const guardIndex = body.indexOf("isFileScopedDriveChange(change)"); + const firstSideEffect = body.indexOf("markSourceMissing("); + + expect(guardIndex).toBeGreaterThan(-1); + expect(firstSideEffect).toBeGreaterThan(-1); + // The skip must come first: nothing downstream can act on a fileId-less + // entry, and the reconcile loop still advances the cursor afterwards. + expect(guardIndex).toBeLessThan(firstSideEffect); + }); + + test("propagates the removal signal instead of swallowing it", () => { + const body = source.slice( + source.indexOf("async function processGoogleDriveChange("), + ); + // A drive removal must return true so reconcileChanges rebuilds the + // selection snapshot; returning a bare false here would strand every + // source imported from the drive that went away. + expect(body).toContain("isDriveRemovalChange(change)"); + expect(body).toContain("return driveRemoved;"); + }); + + test("the reconcile loop persists the cursor on both skip and removal", () => { + const loop = source.slice( + source.indexOf("async function reconcileChanges("), + source.indexOf("async function markConnectorAccessLost("), + ); + // Skipped entry: returns false, requiresSelectionSnapshot stays false, so + // the per-page persist runs. + expect(loop).toContain("if (!requiresSelectionSnapshot) {"); + // Removal: returns true, so the cursor is persisted after the snapshot. + expect(loop).toContain("if (requiresSelectionSnapshot) {"); + expect(loop).toContain("await reconcileSelectionSnapshot("); + // Either way the cursor advances — that is the invariant the outage broke. + expect(loop.match(/await persistSyncCursor\(/g)).toHaveLength(2); + }); +}); diff --git a/tests/unit/lib/repositories/google-drive-changes-feed.test.ts b/tests/unit/lib/repositories/google-drive-changes-feed.test.ts new file mode 100644 index 000000000..4c74a5fe3 --- /dev/null +++ b/tests/unit/lib/repositories/google-drive-changes-feed.test.ts @@ -0,0 +1,89 @@ +/** @jest-environment node */ + +import { GoogleDriveClient } from "@/lib/repositories/google-drive/drive-client"; + +function jsonResponse(body: unknown, status = 200): Response { + return { + ok: status >= 200 && status < 300, + status, + headers: new Headers({ "content-type": "application/json" }), + json: async () => body, + } as unknown as Response; +} + +describe("GoogleDriveClient changes feed", () => { + test("accepts Shared Drive-scoped change entries that carry no fileId", async () => { + // Regression: Google emits changeType "drive" entries (Shared Drive + // rename/membership/restriction events) with no fileId. Requiring fileId + // made listChanges throw before the cursor advanced, so the sync Lambda + // retried the same page forever and the queue filled with poison messages. + const fetchMock = jest + .fn, Parameters>() + .mockResolvedValue( + jsonResponse({ + newStartPageToken: "start-2", + changes: [ + { + changeType: "drive", + time: "2026-08-06T12:00:00.000Z", + driveId: "drive-1", + }, + { + changeType: "file", + fileId: "file-1", + time: "2026-08-06T12:00:01.000Z", + }, + ], + }), + ); + + const page = await new GoogleDriveClient("token", { + fetch: fetchMock, + }).listChanges("cursor-1", "drive-1"); + + expect(page.values).toHaveLength(2); + expect(page.values[0]).toEqual( + expect.objectContaining({ changeType: "drive", driveId: "drive-1" }), + ); + expect(page.values[0]?.fileId).toBeUndefined(); + expect(page.values[1]).toEqual( + expect.objectContaining({ changeType: "file", fileId: "file-1" }), + ); + expect(page.newStartPageToken).toBe("start-2"); + }); + + test("tolerates change types Google has not shipped yet", async () => { + const fetchMock = jest + .fn, Parameters>() + .mockResolvedValue( + jsonResponse({ + newStartPageToken: "start-3", + changes: [{ changeType: "someFutureScope" }], + }), + ); + + const page = await new GoogleDriveClient("token", { + fetch: fetchMock, + }).listChanges("cursor-1"); + + expect(page.values).toEqual([ + expect.objectContaining({ changeType: "someFutureScope" }), + ]); + }); + + test("requests changeType in the changes fields projection", async () => { + const fetchMock = jest + .fn, Parameters>() + .mockResolvedValue(jsonResponse({ newStartPageToken: "start-1" })); + + await new GoogleDriveClient("token", { fetch: fetchMock }).listChanges( + "cursor-1", + ); + + const fields = new URL( + String(fetchMock.mock.calls[0]?.[0]), + ).searchParams.get("fields"); + expect(fields).toContain("changes(changeType,fileId,"); + }); + +});