From fbe8a76df50395b9585b583a5b9ff6b408aa919a Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 16 Sep 2026 19:17:01 -0700 Subject: [PATCH] fix(gmail): open search results in the indexed mailbox --- .../connectors/gmail/company-crawl.test.ts | 13 +- apps/sim/connectors/gmail/gmail.test.ts | 12 +- apps/sim/connectors/gmail/gmail.ts | 58 ++++----- apps/sim/connectors/gmail/mailbox.test.ts | 112 ++++++++++++++++++ apps/sim/connectors/gmail/mailbox.ts | 78 ++++++++++++ .../gmail-member.integration.ts | 17 ++- .../sim/lib/knowledge/search/citation.test.ts | 6 + 7 files changed, 255 insertions(+), 41 deletions(-) create mode 100644 apps/sim/connectors/gmail/mailbox.test.ts create mode 100644 apps/sim/connectors/gmail/mailbox.ts diff --git a/apps/sim/connectors/gmail/company-crawl.test.ts b/apps/sim/connectors/gmail/company-crawl.test.ts index 2bf06db0956..4e73f8ef405 100644 --- a/apps/sim/connectors/gmail/company-crawl.test.ts +++ b/apps/sim/connectors/gmail/company-crawl.test.ts @@ -64,7 +64,11 @@ function providerResponse(url: string, init?: RequestInit): Response { const parsed = new URL(url) const token = new Headers(init?.headers).get('Authorization') const mailbox = token?.includes(BOB.email) ? 'Bob' : 'Alice' - if (parsed.pathname.endsWith('/profile')) return Response.json({ emailAddress: ALICE.email }) + if (parsed.pathname.endsWith('/profile')) + return Response.json({ + emailAddress: mailbox === 'Bob' ? BOB.email : ALICE.email, + historyId: '100', + }) if (parsed.pathname.endsWith('/labels')) { return Response.json({ labels: [{ id: 'Label_7', name: `${mailbox} label` }] }) } @@ -234,6 +238,7 @@ describe('company-wide Gmail indexing', () => { const context = centralContext() const first = await gmailConnector.listDocuments('directory-token', CONFIG, undefined, context) const alice = first.documents[0] + expect(new URL(alice.sourceUrl!).searchParams.get('Email')).toBe(ALICE.email) const aliceBody = await gmailConnector.getDocument( 'directory-token', CONFIG, @@ -247,6 +252,7 @@ describe('company-wide Gmail indexing', () => { context ) const bob = second.documents[0] + expect(new URL(bob.sourceUrl!).searchParams.get('Email')).toBe(BOB.email) const bobBody = await gmailConnector.getDocument( 'directory-token', CONFIG, @@ -322,7 +328,10 @@ describe('company-wide Gmail indexing', () => { resumedContext ) expect(resumed.documents).toEqual(first.documents) - expect(new URL(fetchProvider.mock.calls[1][0]).searchParams.get('q')).toBe(firstQuery) + const listings = fetchProvider.mock.calls.filter(([url]) => + new URL(url).pathname.endsWith('/threads') + ) + expect(new URL(listings[1][0]).searchParams.get('q')).toBe(firstQuery) const body = await gmailConnector.getDocument( 'directory-token', CONFIG, diff --git a/apps/sim/connectors/gmail/gmail.test.ts b/apps/sim/connectors/gmail/gmail.test.ts index 58595617115..b29cc901f07 100644 --- a/apps/sim/connectors/gmail/gmail.test.ts +++ b/apps/sim/connectors/gmail/gmail.test.ts @@ -17,6 +17,12 @@ vi.mock('@/lib/knowledge/documents/secure-fetch.server', () => ({ ? options.fetcher(url, init, mockFetchWithRetry) : mockFetchWithRetry(url, init), })) +vi.mock('@/connectors/gmail/mailbox', async (importOriginal) => ({ + ...(await importOriginal()), + getGmailMailboxEmail: vi.fn(async (token: string) => + token === 'bob-token' ? 'bob@example.com' : 'alice@example.com' + ), +})) vi.mock('@/components/icons', () => ({ GmailIcon: () => null })) vi.mock('@/lib/knowledge/documents/service', () => ({ isTriggerAvailable: () => false, @@ -789,7 +795,8 @@ describe('Gmail Search member isolation', () => { externalId: 'member:alice:thread-1', contentHash: 'gmail:thread-1:10:body-v2', contentDeferred: false, - sourceUrl: 'https://mail.google.com/mail/u/0/#all/thread-1', + sourceUrl: + 'https://accounts.google.com/AccountChooser?Email=alice%40example.com&continue=https%3A%2F%2Fmail.google.com%2Fmail%2F%3Fauthuser%3Dalice%2540example.com%23all%2Fthread-1', }) expect(document?.content).toContain('Private mailbox content') expect(mockFetchWithRetry.mock.calls[0][0]).toContain('/threads/thread-1?format=full') @@ -1221,7 +1228,8 @@ describe('Gmail change feed', () => { mockFetchWithRetry.mockImplementation(async (url: string) => { const parsed = new URL(url) requests.push(parsed) - if (parsed.pathname.endsWith('/profile')) return Response.json({ historyId: '500' }) + if (parsed.pathname.endsWith('/profile')) + return Response.json({ emailAddress: 'alice@example.com', historyId: '500' }) if (parsed.pathname.endsWith('/labels')) return Response.json({ labels }) if (parsed.pathname.endsWith('/history')) { return Response.json(pages[historyCall++] ?? historyPage([])) diff --git a/apps/sim/connectors/gmail/gmail.ts b/apps/sim/connectors/gmail/gmail.ts index 4f2be5cdde1..2063e8e4aa2 100644 --- a/apps/sim/connectors/gmail/gmail.ts +++ b/apps/sim/connectors/gmail/gmail.ts @@ -4,6 +4,7 @@ import { isPlainRecord } from '@sim/utils/object' import { mapWithConcurrency } from '@/lib/core/utils/concurrency' import { isPayloadSizeLimitError, readResponseJsonWithLimit } from '@/lib/core/utils/stream-limits' import { VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/documents/utils' +import { getGmailMailboxEmail, getGmailProfile, gmailThreadUrl } from '@/connectors/gmail/mailbox' import { DEFAULT_MAX_THREADS, gmailConnectorMeta } from '@/connectors/gmail/meta' import { fetchGoogleApiWithRetry, GoogleApiError } from '@/connectors/google-workspace/api-errors' import { @@ -621,10 +622,12 @@ async function resolveLabelNames( * Creates a lightweight document stub from a thread list entry. * Uses metadata-based contentHash for change detection without downloading content. */ -function threadToStub( +async function threadToStub( thread: GmailThread, + accessToken: string, syncContext?: Record -): ExternalDocument { +): Promise { + const mailboxEmail = await getGmailMailboxEmail(accessToken, syncContext) return { externalId: memberDocumentId(thread.id, syncContext), title: thread.snippet || 'Untitled Thread', @@ -632,7 +635,7 @@ function threadToStub( contentDeferred: true, estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, mimeType: 'text/plain', - sourceUrl: threadUrl(thread.id), + sourceUrl: gmailThreadUrl(thread.id, mailboxEmail), /** Rehydrate older rows that omitted separately stored message bodies. */ contentHash: `gmail:${thread.id}:${thread.historyId}:body-v2`, skippedRetryPolicy: 'source-change', @@ -640,15 +643,6 @@ function threadToStub( } } -/** - * Deep link to a thread. `#all` is used rather than `#inbox` because a synced - * thread may be archived or live only under a user label, where an `#inbox` - * fragment resolves to nothing. - */ -function threadUrl(threadId: string): string { - return `https://mail.google.com/mail/u/0/#all/${threadId}` -} - /** A feed position: the mailbox history id the next read starts from, mid-page when paging. */ interface GmailChangeCursor { historyId: string @@ -780,22 +774,7 @@ const gmailMailboxConnector: ConnectorConfig = { if (syncContext?.mirrorsSourceAcls === true) { throw new Error('Company-wide Gmail indexing uses complete mailbox listings') } - const response = await fetchGoogleApiWithRetry( - 'gmail.users.getProfile', - `${GMAIL_API_BASE}/profile?fields=historyId`, - { - method: 'GET', - headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json' }, - } - ) - const data: unknown = await response.json() - if ( - !isPlainRecord(data) || - typeof data.historyId !== 'string' || - !/^\d+$/.test(data.historyId) - ) { - throw new Error('Gmail returned malformed profile metadata') - } + const data = await getGmailProfile(accessToken, syncContext) return JSON.stringify({ historyId: data.historyId }) }, @@ -813,7 +792,7 @@ const gmailMailboxConnector: ConnectorConfig = { accessToken: string, sourceConfig: Record, cursor: string, - syncContext?: Record + syncContext: Record = {} ): Promise => { if (syncContext?.mirrorsSourceAcls === true) { throw new Error('Company-wide Gmail indexing uses complete mailbox listings') @@ -854,7 +833,11 @@ const gmailMailboxConnector: ConnectorConfig = { const externalId = memberDocumentId(threadId, syncContext) const thread = await fetchThread(accessToken, threadId, 'metadata') if (!thread || !threadInScope(thread, scope)) return { kind: 'removed', externalId } - return { kind: 'upsert', externalId, document: threadToStub(thread, syncContext) } + return { + kind: 'upsert', + externalId, + document: await threadToStub(thread, accessToken, syncContext), + } } ) @@ -873,7 +856,7 @@ const gmailMailboxConnector: ConnectorConfig = { accessToken: string, sourceConfig: Record, cursor?: string, - syncContext?: Record + syncContext: Record = {} ): Promise => { const signal = syncContext?.signal instanceof AbortSignal ? syncContext.signal : undefined signal?.throwIfAborted() @@ -962,7 +945,7 @@ const gmailMailboxConnector: ConnectorConfig = { const metadata = thread.historyId ? thread : await fetchThread(accessToken, thread.id, 'minimal', signal) - return metadata ? threadToStub(metadata, syncContext) : null + return metadata ? threadToStub(metadata, accessToken, syncContext) : null }) const documents = stubs.filter((stub): stub is ExternalDocument => stub !== null) @@ -1002,7 +985,7 @@ const gmailMailboxConnector: ConnectorConfig = { accessToken: string, _sourceConfig: Record, externalId: string, - syncContext?: Record + syncContext: Record = {} ): Promise => { const signal = syncContext?.signal instanceof AbortSignal ? syncContext.signal : undefined signal?.throwIfAborted() @@ -1028,7 +1011,7 @@ const gmailMailboxConnector: ConnectorConfig = { } return { ...markSkipped( - threadToStub(after, syncContext), + await threadToStub(after, accessToken, syncContext), sizeLimitSkipReason(MAX_THREAD_RESPONSE_BYTES) ), skippedExistingDisposition: 'replace', @@ -1043,7 +1026,10 @@ const gmailMailboxConnector: ConnectorConfig = { } catch (error) { if (error instanceof ConnectorFileTooLargeError) { return { - ...markSkipped(threadToStub(thread, syncContext), sizeLimitSkipReason(error.limitBytes)), + ...markSkipped( + await threadToStub(thread, accessToken, syncContext), + sizeLimitSkipReason(error.limitBytes) + ), skippedExistingDisposition: 'replace', } } @@ -1056,7 +1042,7 @@ const gmailMailboxConnector: ConnectorConfig = { metadata.labels = await resolveLabelNames(accessToken, labelIds, syncContext) return { - ...threadToStub(thread, syncContext), + ...(await threadToStub(thread, accessToken, syncContext)), title: subject, content, contentDeferred: false, diff --git a/apps/sim/connectors/gmail/mailbox.test.ts b/apps/sim/connectors/gmail/mailbox.test.ts new file mode 100644 index 00000000000..8003be4fef0 --- /dev/null +++ b/apps/sim/connectors/gmail/mailbox.test.ts @@ -0,0 +1,112 @@ +/** @vitest-environment node */ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { getGmailMailboxEmail, getGmailProfile, gmailThreadUrl } from '@/connectors/gmail/mailbox' + +const { transport } = vi.hoisted(() => ({ transport: vi.fn() })) +vi.mock('@/lib/knowledge/documents/secure-fetch.server', () => ({ + fetchWithRetry: ( + url: string, + init: RequestInit, + options: { + fetcher: (url: string, init: RequestInit, transport: typeof fetch) => Promise + } + ) => options.fetcher(url, init, transport), +})) + +beforeEach(() => { + transport.mockReset() + transport.mockResolvedValue( + Response.json({ emailAddress: 'Alice+work@example.com', historyId: '123' }) + ) +}) + +describe('Gmail mailbox identity', () => { + it('selects the authenticated mailbox before an archived-thread fragment', async () => { + const email = await getGmailMailboxEmail('mailbox-token', {}) + const url = new URL(gmailThreadUrl('19a3f0123456789', email)) + expect(url.origin).toBe('https://accounts.google.com') + expect(url.pathname).toBe('/AccountChooser') + expect(url.searchParams.get('Email')).toBe('alice+work@example.com') + expect(url.hash).toBe('') + const destination = new URL(url.searchParams.get('continue')!) + expect(destination.origin).toBe('https://mail.google.com') + expect(destination.pathname).toBe('/mail/') + expect(destination.searchParams.get('authuser')).toBe('alice+work@example.com') + expect(destination.hash).toBe('#all/19a3f0123456789') + expect(transport).toHaveBeenCalledWith( + 'https://gmail.googleapis.com/gmail/v1/users/me/profile?fields=emailAddress,historyId', + expect.objectContaining({ + headers: expect.objectContaining({ Authorization: 'Bearer mailbox-token' }), + }) + ) + }) + + it('deduplicates concurrent lookups and reuses the profile read for the history watermark', async () => { + const context = {} + expect( + await Promise.all(Array.from({ length: 100 }, () => getGmailMailboxEmail('token', context))) + ).toEqual(Array(100).fill('alice+work@example.com')) + expect(transport).toHaveBeenCalledTimes(1) + transport.mockResolvedValue( + Response.json({ emailAddress: 'alice+work@example.com', historyId: '456' }) + ) + expect((await getGmailProfile('token', context)).historyId).toBe('456') + expect(await getGmailMailboxEmail('token', context)).toBe('alice+work@example.com') + expect(transport).toHaveBeenCalledTimes(2) + }) + + it('seeds mailbox identity from a fresh watermark without a second request', async () => { + const context = {} + await getGmailProfile('token', context) + await getGmailMailboxEmail('token', context) + expect(transport).toHaveBeenCalledTimes(1) + }) + + it('isolates credentials even if the same context is mistakenly reused', async () => { + const context = {} + await getGmailMailboxEmail('alice-token', context) + transport.mockImplementation(async () => + Response.json({ emailAddress: 'bob@example.com', historyId: '321' }) + ) + expect(await getGmailMailboxEmail('bob-token', context)).toBe('bob@example.com') + await getGmailMailboxEmail('bob-token', {}) + expect(transport).toHaveBeenCalledTimes(3) + }) + + it.each([ + {}, + { emailAddress: '', historyId: '1' }, + { emailAddress: 'not-an-email', historyId: '1' }, + { emailAddress: 'alice@example.com', historyId: 123 }, + { emailAddress: 'alice@example.com', historyId: 'invalid' }, + ])('rejects malformed profiles instead of falling back to account zero: %j', async (body) => { + transport.mockResolvedValue(Response.json(body)) + await expect(getGmailMailboxEmail('token', {})).rejects.toThrow('malformed profile') + }) + + it('does not cache a failed lookup', async () => { + transport.mockResolvedValueOnce(Response.json({})) + const context = {} + await expect(getGmailMailboxEmail('token', context)).rejects.toThrow() + expect(await getGmailMailboxEmail('token', context)).toBe('alice+work@example.com') + }) + + it('preserves provider authentication errors and cancellation', async () => { + transport.mockResolvedValueOnce(new Response(null, { status: 401 })) + await expect(getGmailMailboxEmail('invalid', {})).rejects.toMatchObject({ status: 401 }) + const controller = new AbortController() + controller.abort() + await expect(getGmailProfile('token', { signal: controller.signal })).rejects.toThrow() + expect(transport).toHaveBeenCalledTimes(1) + }) + + it('bounds profile responses', async () => { + transport.mockResolvedValue(Response.json({ padding: 'x'.repeat(17 * 1024) })) + await expect(getGmailMailboxEmail('token', {})).rejects.toThrow() + }) + + it('encodes fragment delimiters rather than allowing them to change the link target', () => { + const chooser = new URL(gmailThreadUrl('thread/?#id', 'alice@example.com')) + expect(new URL(chooser.searchParams.get('continue')!).hash).toBe('#all/thread%2F%3F%23id') + }) +}) diff --git a/apps/sim/connectors/gmail/mailbox.ts b/apps/sim/connectors/gmail/mailbox.ts new file mode 100644 index 00000000000..1e3a087b0a9 --- /dev/null +++ b/apps/sim/connectors/gmail/mailbox.ts @@ -0,0 +1,78 @@ +import { isPlainRecord } from '@sim/utils/object' +import { isValidEmailSyntax, normalizeEmail } from '@sim/utils/string' +import { readResponseJsonWithLimit } from '@/lib/core/utils/stream-limits' +import { fetchGoogleApiWithRetry } from '@/connectors/google-workspace/api-errors' + +interface GmailProfile { + emailAddress: string + historyId: string +} + +/** Sync-local only: never persist tokens or share mailbox identity across credentials. */ +const mailboxes = new WeakMap< + Record, + { accessToken: string; email: Promise } +>() + +/** Always reads a fresh history watermark; only mailbox identity is reused within a sync. */ +export async function getGmailProfile( + accessToken: string, + syncContext?: Record +): Promise { + const signal = syncContext?.signal instanceof AbortSignal ? syncContext.signal : undefined + signal?.throwIfAborted() + const response = await fetchGoogleApiWithRetry( + 'gmail.users.getProfile', + 'https://gmail.googleapis.com/gmail/v1/users/me/profile?fields=emailAddress,historyId', + { + method: 'GET', + signal, + headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json' }, + } + ) + const data = await readResponseJsonWithLimit(response, { + maxBytes: 16 * 1024, + label: 'Gmail profile', + }) + if ( + !isPlainRecord(data) || + typeof data.emailAddress !== 'string' || + !isValidEmailSyntax(data.emailAddress) || + typeof data.historyId !== 'string' || + !/^\d+$/.test(data.historyId) + ) { + throw new Error('Gmail returned malformed profile metadata') + } + const emailAddress = normalizeEmail(data.emailAddress) + if (syncContext) { + mailboxes.set(syncContext, { accessToken, email: Promise.resolve(emailAddress) }) + } + return { emailAddress, historyId: data.historyId } +} + +export function getGmailMailboxEmail( + accessToken: string, + syncContext?: Record +): Promise { + const cached = syncContext && mailboxes.get(syncContext) + if (cached?.accessToken === accessToken) return cached.email + const email = getGmailProfile(accessToken, syncContext).then((profile) => profile.emailAddress) + if (syncContext) { + mailboxes.set(syncContext, { accessToken, email }) + void email.catch(() => { + if (mailboxes.get(syncContext)?.email === email) mailboxes.delete(syncContext) + }) + } + return email +} + +/** The chooser handles signed-out mailboxes; authuser alone can fall back to account zero. */ +export function gmailThreadUrl(threadId: string, mailboxEmail: string): string { + const thread = new URL('https://mail.google.com/mail/') + thread.searchParams.set('authuser', mailboxEmail) + thread.hash = `all/${encodeURIComponent(threadId)}` + const chooser = new URL('https://accounts.google.com/AccountChooser') + chooser.searchParams.set('Email', mailboxEmail) + chooser.searchParams.set('continue', thread.toString()) + return chooser.toString() +} diff --git a/apps/sim/lib/knowledge/__integration__/gmail-member.integration.ts b/apps/sim/lib/knowledge/__integration__/gmail-member.integration.ts index a3abd7da45f..02bebeabca4 100644 --- a/apps/sim/lib/knowledge/__integration__/gmail-member.integration.ts +++ b/apps/sim/lib/knowledge/__integration__/gmail-member.integration.ts @@ -114,6 +114,11 @@ describe('Gmail member ingestion and ACLs in PostgreSQL (provider fixtures)', () Response.json({ error: { code: 401, message: 'Invalid Credentials' } }, { status: 401 }) ) } + if (url.pathname === '/gmail/v1/users/me/profile') { + return Promise.resolve( + Response.json({ emailAddress: `mailbox-${member}@example.com`, historyId: '900' }) + ) + } if (url.pathname === '/gmail/v1/users/me/labels') { return Promise.resolve( Response.json({ labels: [{ id: 'INBOX', name: 'INBOX', type: 'system' }] }) @@ -411,7 +416,12 @@ describe('Gmail member ingestion and ACLs in PostgreSQL (provider fixtures)', () expect(new Set(own.map((row) => row.externalId))).toEqual( new Set([`member:${member.id}:shared-thread-id`, `member:${member.id}:private-${index}`]) ) - for (const row of own) expect(row.acl).toEqual([member.subjectToken]) + for (const row of own) { + expect(row.acl).toEqual([member.subjectToken]) + expect(new URL(row.sourceUrl!).searchParams.get('Email')).toBe( + `mailbox-${index}@example.com` + ) + } const observations = await db .select() .from(knowledgeDocumentObservation) @@ -424,6 +434,11 @@ describe('Gmail member ingestion and ACLs in PostgreSQL (provider fixtures)', () expect(new Set(results.map((row) => row.documentId))).toEqual( new Set(own.map((row) => row.id)) ) + for (const result of results) { + expect(new URL(result.sourceUrl!).searchParams.get('Email')).toBe( + `mailbox-${index}@example.com` + ) + } expect(results.map((row) => row.content).join('\n')).toContain( index === 0 ? 'Alice private reply' : 'Bob private reply' ) diff --git a/apps/sim/lib/knowledge/search/citation.test.ts b/apps/sim/lib/knowledge/search/citation.test.ts index 982760bce31..e5977bff14c 100644 --- a/apps/sim/lib/knowledge/search/citation.test.ts +++ b/apps/sim/lib/knowledge/search/citation.test.ts @@ -21,6 +21,12 @@ describe('knowledge citations', () => { }) }) + it('preserves Gmail mailbox selection and thread targeting', () => { + const sourceUrl = + 'https://accounts.google.com/AccountChooser?Email=alice%2Bwork%40example.com&continue=https%3A%2F%2Fmail.google.com%2Fmail%2F%3Fauthuser%3Dalice%252Bwork%2540example.com%23all%2F19a3f0123456789' + expect(createKnowledgeDocumentCitation({ ...input, sourceUrl }).citationUrl).toBe(sourceUrl) + }) + it.each([ null, '',