Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions apps/sim/connectors/gmail/company-crawl.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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` }] })
}
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
12 changes: 10 additions & 2 deletions apps/sim/connectors/gmail/gmail.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<typeof import('@/connectors/gmail/mailbox')>()),
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,
Expand Down Expand Up @@ -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')
Expand Down Expand Up @@ -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([]))
Expand Down
58 changes: 22 additions & 36 deletions apps/sim/connectors/gmail/gmail.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -621,34 +622,27 @@ 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<string, unknown>
): ExternalDocument {
): Promise<ExternalDocument> {
const mailboxEmail = await getGmailMailboxEmail(accessToken, syncContext)
return {
externalId: memberDocumentId(thread.id, syncContext),
title: thread.snippet || 'Untitled Thread',
content: '',
contentDeferred: true,
estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES,
mimeType: 'text/plain',
sourceUrl: threadUrl(thread.id),
sourceUrl: gmailThreadUrl(thread.id, mailboxEmail),
Comment thread
waleedlatif1 marked this conversation as resolved.
/** Rehydrate older rows that omitted separately stored message bodies. */
contentHash: `gmail:${thread.id}:${thread.historyId}:body-v2`,
skippedRetryPolicy: 'source-change',
metadata: {},
}
}

/**
* 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
Expand Down Expand Up @@ -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 })
},

Expand All @@ -813,7 +792,7 @@ const gmailMailboxConnector: ConnectorConfig = {
accessToken: string,
sourceConfig: Record<string, unknown>,
cursor: string,
syncContext?: Record<string, unknown>
syncContext: Record<string, unknown> = {}
): Promise<ExternalChangeList> => {
if (syncContext?.mirrorsSourceAcls === true) {
throw new Error('Company-wide Gmail indexing uses complete mailbox listings')
Expand Down Expand Up @@ -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),
}
}
)

Expand All @@ -873,7 +856,7 @@ const gmailMailboxConnector: ConnectorConfig = {
accessToken: string,
sourceConfig: Record<string, unknown>,
cursor?: string,
syncContext?: Record<string, unknown>
syncContext: Record<string, unknown> = {}
): Promise<ExternalDocumentList> => {
const signal = syncContext?.signal instanceof AbortSignal ? syncContext.signal : undefined
signal?.throwIfAborted()
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -1002,7 +985,7 @@ const gmailMailboxConnector: ConnectorConfig = {
accessToken: string,
_sourceConfig: Record<string, unknown>,
externalId: string,
syncContext?: Record<string, unknown>
syncContext: Record<string, unknown> = {}
): Promise<ExternalDocument | null> => {
const signal = syncContext?.signal instanceof AbortSignal ? syncContext.signal : undefined
signal?.throwIfAborted()
Expand All @@ -1028,7 +1011,7 @@ const gmailMailboxConnector: ConnectorConfig = {
}
return {
...markSkipped(
threadToStub(after, syncContext),
await threadToStub(after, accessToken, syncContext),
sizeLimitSkipReason(MAX_THREAD_RESPONSE_BYTES)
),
skippedExistingDisposition: 'replace',
Expand All @@ -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',
}
}
Expand All @@ -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,
Expand Down
112 changes: 112 additions & 0 deletions apps/sim/connectors/gmail/mailbox.test.ts
Original file line number Diff line number Diff line change
@@ -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<typeof fetch>() }))
vi.mock('@/lib/knowledge/documents/secure-fetch.server', () => ({
fetchWithRetry: (
url: string,
init: RequestInit,
options: {
fetcher: (url: string, init: RequestInit, transport: typeof fetch) => Promise<Response>
}
) => 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')
})
})
Loading
Loading