fix(api): hold the gmail sync cursor when message fetches fail

apps/api/src/google/gmail-sync.service.ts

@@ -10,6 +10,7 @@ import { Injectable, Logger } from "@nestjs/common";
 import { ActivityStampService } from "../crm/activity-stamp.service";
 import { InjectDatabase } from "../database/database.constants";
 import { GmailClient, type GmailMessage } from "./gmail.client";
+import type { GoogleResult } from "./google-api.client";
 import { GoogleMatchService, type MatchContext } from "./google-match.service";
 import { GoogleTokenService } from "./google-token.service";
 import {
@@ -169,7 +170,7 @@ export class GmailSyncService {
 			}
 		}
 
-		const { written, remaining } = await this.ingest(
+		const { written, remaining, deferred } = await this.ingest(
 			row,
 			accessToken,
 			mailbox,
@@ -178,18 +179,19 @@ export class GmailSyncService {
 
 		await this.state.settle(row.id, {
 			cursor:
-				remaining > 0
+				remaining > 0 || deferred > 0
 					? startHistoryId
 					: (history.data.historyId ?? startHistoryId),
 			status: GoogleSyncStatus.RUNNING,
 		});
 
-		if (written > 0 || remaining > 0) {
+		if (written > 0 || remaining > 0 || deferred > 0) {
 			this.logger.log({
 				message: "Gmail incremental sync",
 				userId: row.userId,
 				messagesWritten: written,
 				remaining,
+				deferred,
 			});
 		}
 
@@ -206,8 +208,8 @@ export class GmailSyncService {
 		accessToken: string,
 		mailbox: string,
 		ids: readonly string[],
-	): Promise<{ written: number; remaining: number }> {
-		if (ids.length === 0) return { written: 0, remaining: 0 };
+	): Promise<{ written: number; remaining: number; deferred: number }> {
+		if (ids.length === 0) return { written: 0, remaining: 0, deferred: 0 };
 
 		const alreadyHave = await this.db.emailMessage.findMany({
 			where: { gmailMessageId: { in: [...ids] } },
@@ -221,7 +223,7 @@ export class GmailSyncService {
 		const batch = pending.slice(0, MAX_MESSAGES_PER_TICK);
 		const remaining = pending.length - batch.length;
 
-		if (batch.length === 0) return { written: 0, remaining };
+		if (batch.length === 0) return { written: 0, remaining, deferred: 0 };
 
 		const [internal, suppressedDomains, suppressedEmails] = await Promise.all([
 			this.match.internalIdentity(),
@@ -237,16 +239,20 @@ export class GmailSyncService {
 		};
 
 		let written = 0;
+		let deferred = 0;
 
 		for (const id of batch) {
 			const message = await this.gmail.getMessage(accessToken, id);
-			if (message.outcome !== "ok") continue;
+			if (message.outcome !== "ok") {
+				if (isRetryable(message)) deferred += 1;
+				continue;
+			}
 
 			const stored = await this.store(row, mailbox, message.data, context);
 			if (stored) written += 1;
 		}
 
-		return { written, remaining };
+		return { written, remaining, deferred };
 	}
 
 	private async store(
@@ -502,3 +508,10 @@ export class GmailSyncService {
 		};
 	}
 }
+
+function isRetryable(
+	result: GoogleResult<unknown>,
+): result is Extract<GoogleResult<unknown>, { outcome: "failed" }> {
+	if (result.outcome === "failed") return result.retryable;
+	return result.outcome === "rate-limited" || result.outcome === "unauthorized";
+}

apps/api/test/gmail-sync.spec.ts

@@ -0,0 +1,192 @@
+import { beforeEach, describe, expect, it } from "bun:test";
+import type { Db, MailboxSyncModel } from "@crm/db";
+import { GoogleSyncStatus } from "@crm/db";
+import type { ActivityStampService } from "../src/crm/activity-stamp.service";
+import type { GmailClient } from "../src/google/gmail.client";
+import { GmailSyncService } from "../src/google/gmail-sync.service";
+import type { GoogleMatchService } from "../src/google/google-match.service";
+import type { GoogleTokenService } from "../src/google/google-token.service";
+import type { SyncStateService } from "../src/google/sync-state.service";
+
+function harness(messageOutcomes: unknown[]) {
+	const settled: Array<{ cursor?: string | null }> = [];
+	let getMessageIndex = 0;
+
+	const db = {
+		emailMessage: {
+			findMany: async () => [],
+			findUnique: async () => null,
+			aggregate: async () => ({
+				_count: { _all: 0 },
+				_min: { sentAt: null },
+				_max: { sentAt: null },
+			}),
+			create: async () => ({}),
+		},
+		emailThread: {
+			findUnique: async () => null,
+			findFirst: async () => null,
+			upsert: async () => ({
+				id: "thread-1",
+				firstMessageAt: new Date(),
+				lastMessageAt: new Date(),
+			}),
+			update: async () => ({}),
+		},
+		activity: {
+			upsert: async () => ({ createdAt: new Date() }),
+		},
+		mailboxSync: {
+			update: async ({ data }: { data: { cursor?: string | null } }) => {
+				settled.push({ cursor: data.cursor ?? null });
+				return {};
+			},
+		},
+	} as unknown as Db;
+
+	const service = new GmailSyncService(
+		db,
+		{
+			profile: async () => ({
+				outcome: "ok" as const,
+				data: { emailAddress: "rep@acme.com" },
+			}),
+			listHistory: async () => ({
+				outcome: "ok" as const,
+				data: {
+					historyId: "200",
+					history: [
+						{ messagesAdded: [{ message: { id: "m1" } }] },
+						{ messagesAdded: [{ message: { id: "m2" } }] },
+					],
+				},
+			}),
+			getMessage: async () => {
+				const outcome = messageOutcomes[getMessageIndex] ?? {
+					outcome: "failed",
+					reason: "unexpected",
+					retryable: false,
+				};
+				getMessageIndex += 1;
+				return outcome;
+			},
+		} as unknown as GmailClient,
+		{
+			accessTokenFor: async () => ({
+				outcome: "ok" as const,
+				accessToken: "token",
+			}),
+		} as unknown as GoogleTokenService,
+		{
+			internalIdentity: async () => ({
+				domains: new Set(["acme.com"]),
+				addresses: new Set(["rep@acme.com"]),
+			}),
+			suppressedDomains: async () => new Set<string>(),
+			suppressedEmails: async () => new Set<string>(),
+		} as unknown as GoogleMatchService,
+		{
+			settle: async (_id: string, update: { cursor?: string | null }) => {
+				settled.push({ cursor: update.cursor ?? null });
+			},
+			markRunning: async () => undefined,
+			markFailed: async () => undefined,
+			markNeedsReconnect: async () => undefined,
+			markRateLimited: async () => undefined,
+			clearCursor: async () => undefined,
+		} as unknown as SyncStateService,
+		{
+			touch: async () => undefined,
+		} as unknown as ActivityStampService,
+	);
+
+	return { service, settled };
+}
+
+const row = {
+	id: "sync-1",
+	userId: "u1",
+	source: "gmail",
+	status: GoogleSyncStatus.RUNNING,
+	cursor: "100",
+	lastSyncedAt: new Date(),
+	lastError: null,
+	retryAfter: null,
+	autoCreate: false,
+	createdAt: new Date(),
+	updatedAt: new Date(),
+} as MailboxSyncModel;
+
+describe("GmailSyncService cursor", () => {
+	beforeEach(() => {
+		process.env.ALLOWED_SIGN_IN = "acme.com";
+	});
+
+	it("holds the cursor at the start history id when a message fetch is rate-limited", async () => {
+		const { service, settled } = harness([
+			{
+				outcome: "rate-limited",
+				reason: "User Rate Limit Exceeded",
+				retryAfterMs: 60_000,
+			},
+			{ outcome: "ok", data: {} },
+		]);
+
+		await service.sync(row);
+
+		expect(settled).toHaveLength(1);
+		expect(settled[0]?.cursor).toBe("100");
+	});
+
+	it("holds the cursor when a message fetch times out (retryable failure)", async () => {
+		const { service, settled } = harness([
+			{
+				outcome: "failed",
+				reason: "Timed out after 20000ms.",
+				retryable: true,
+			},
+			{ outcome: "ok", data: {} },
+		]);
+
+		await service.sync(row);
+
+		expect(settled).toHaveLength(1);
+		expect(settled[0]?.cursor).toBe("100");
+	});
+
+	it("advances the cursor past a message Gmail reports deleted", async () => {
+		const { service, settled } = harness([
+			{ outcome: "cursor-invalid", reason: "Requested entity was not found." },
+			{ outcome: "ok", data: {} },
+		]);
+
+		await service.sync(row);
+
+		expect(settled).toHaveLength(1);
+		expect(settled[0]?.cursor).not.toBe("100");
+	});
+
+	it("advances the cursor past a non-retryable failure", async () => {
+		const { service, settled } = harness([
+			{ outcome: "failed", reason: "Forbidden", retryable: false },
+			{ outcome: "ok", data: {} },
+		]);
+
+		await service.sync(row);
+
+		expect(settled).toHaveLength(1);
+		expect(settled[0]?.cursor).not.toBe("100");
+	});
+
+	it("advances the cursor when every message is fetched", async () => {
+		const { service, settled } = harness([
+			{ outcome: "ok", data: {} },
+			{ outcome: "ok", data: {} },
+		]);
+
+		await service.sync(row);
+
+		expect(settled).toHaveLength(1);
+		expect(settled[0]?.cursor).not.toBe("100");
+	});
+});