diff --git a/apps/api/src/google/gmail-sync.service.ts b/apps/api/src/google/gmail-sync.service.ts index eccfcc92..684313d7 100644 --- a/apps/api/src/google/gmail-sync.service.ts +++ b/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, +): result is Extract, { outcome: "failed" }> { + if (result.outcome === "failed") return result.retryable; + return result.outcome === "rate-limited" || result.outcome === "unauthorized"; +} diff --git a/apps/api/test/gmail-sync.spec.ts b/apps/api/test/gmail-sync.spec.ts new file mode 100644 index 00000000..6d04b267 --- /dev/null +++ b/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(), + suppressedEmails: async () => new Set(), + } 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"); + }); +});