import { createLogger } from '@sim/logger' import { getErrorMessage, toError } from '@sim/utils/errors' import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/documents/utils' import { DEFAULT_MAX_THREADS, gmailConnectorMeta } from '@/connectors/gmail/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { htmlToPlainText, joinTagArray, parseMultiValue, parseTagDate } from '@/connectors/utils' const logger = createLogger('GmailConnector') const GMAIL_API_BASE = 'https://gmail.googleapis.com/gmail/v1/users/me' const THREADS_PER_PAGE = 100 interface GmailHeader { name: string value: string } interface GmailMessagePart { mimeType?: string body?: { data?: string; size?: number } parts?: GmailMessagePart[] headers?: GmailHeader[] } interface GmailMessage { id: string threadId: string internalDate?: string payload?: GmailMessagePart labelIds?: string[] snippet?: string } interface GmailThread { id: string historyId?: string messages?: GmailMessage[] snippet?: string } interface GmailLabel { id: string name: string type?: string } /** * Formats a single Gmail label name for use in a `label:` operator. * Gmail search syntax accepts quoted strings for labels containing spaces; * unquoted label tokens have spaces replaced with hyphens. */ function formatLabelToken(name: string): string { const trimmed = name.trim() if (!trimmed) return '' if (/\s/.test(trimmed)) { const escaped = trimmed.replace(/\\/g, '\\\\').replace(/"/g, '\\"') return `label:"${escaped}"` } return `label:${trimmed}` } /** * Builds a Gmail search query string from the source config. * Combines the user's custom query with the label and date range filters. * When multiple labels are provided, they are OR-joined: `(label:A OR label:B)`. */ function buildSearchQuery(sourceConfig: Record): string { const parts: string[] = [] const labelNames = parseMultiValue(sourceConfig.label) if (labelNames.length === 1) { const token = formatLabelToken(labelNames[0]) if (token) parts.push(token) } else if (labelNames.length > 1) { const tokens = labelNames.map(formatLabelToken).filter(Boolean) if (tokens.length === 1) { parts.push(tokens[0]) } else if (tokens.length > 1) { parts.push(`(${tokens.join(' OR ')})`) } } const dateRange = (sourceConfig.dateRange as string) || 'all' const now = new Date() switch (dateRange) { case '7d': parts.push(`after:${formatGmailDate(daysAgo(now, 7))}`) break case '30d': parts.push(`after:${formatGmailDate(daysAgo(now, 30))}`) break case '90d': parts.push(`after:${formatGmailDate(daysAgo(now, 90))}`) break case '6m': parts.push(`after:${formatGmailDate(daysAgo(now, 180))}`) break case '1y': parts.push(`after:${formatGmailDate(daysAgo(now, 365))}`) break } const excludePromotions = sourceConfig.excludePromotions !== 'false' if (excludePromotions) { parts.push('-category:promotions') } const excludeSocial = sourceConfig.excludeSocial !== 'false' if (excludeSocial) { parts.push('-category:social') } const customQuery = sourceConfig.query as string | undefined const trimmedCustom = customQuery?.trim() if (trimmedCustom) { /** * Wrap the user-supplied query in parentheses whenever it contains an OR * so it's AND-joined as a single clause with the preceding label / category * / date filters. Always wrap (rather than try to detect existing outer * parens) because a regex like /^\(.*\)$/ misclassifies inputs such as * `(from:alice) OR (from:bob)` where the parens don't bracket the whole * expression. Double-wrapping is a no-op in Gmail search syntax. */ const needsGroup = /\bOR\b/i.test(trimmedCustom) parts.push(needsGroup ? `(${trimmedCustom})` : trimmedCustom) } return parts.join(' ') } function daysAgo(now: Date, days: number): Date { return new Date(now.getTime() - days * 24 * 60 * 60 * 1000) } function formatGmailDate(date: Date): string { const y = date.getFullYear() const m = String(date.getMonth() + 1).padStart(2, '0') const d = String(date.getDate()).padStart(2, '0') return `${y}/${m}/${d}` } /** * Decodes base64url-encoded content from the Gmail API. * Uses Buffer to correctly handle multi-byte UTF-8 characters. */ function decodeBase64Url(data: string): string { return Buffer.from(data, 'base64url').toString('utf-8') } /** * Extracts the plain text body from a Gmail message payload. * Prefers text/plain, falls back to text/html with tag stripping. */ function extractBody(part: GmailMessagePart): string { if (part.mimeType === 'text/plain' && part.body?.data) { return decodeBase64Url(part.body.data) } if (part.parts) { // Prefer text/plain from multipart for (const child of part.parts) { if (child.mimeType === 'text/plain' && child.body?.data) { return decodeBase64Url(child.body.data) } } // Fall back to text/html for (const child of part.parts) { if (child.mimeType === 'text/html' && child.body?.data) { return htmlToPlainText(decodeBase64Url(child.body.data)) } } // Recurse into nested multipart for (const child of part.parts) { const result = extractBody(child) if (result) return result } } if (part.mimeType === 'text/html' && part.body?.data) { return htmlToPlainText(decodeBase64Url(part.body.data)) } return '' } /** * Gets a header value from a Gmail message payload. */ function getHeader(payload: GmailMessagePart | undefined, name: string): string | undefined { if (!payload?.headers) return undefined const header = payload.headers.find((h) => h.name.toLowerCase() === name.toLowerCase()) return header?.value } /** * Formats a thread's messages into a single document string. */ function formatThread(thread: GmailThread): { content: string subject: string metadata: Record } { const messages = thread.messages || [] if (messages.length === 0) { return { content: '', subject: 'Untitled Thread', metadata: {} } } const firstMessage = messages[0] const lastMessage = messages[messages.length - 1] const subject = getHeader(firstMessage.payload, 'Subject') || 'No Subject' const from = getHeader(firstMessage.payload, 'From') || 'Unknown' const to = getHeader(firstMessage.payload, 'To') || '' const labelIds = firstMessage.labelIds || [] const lines: string[] = [] lines.push(`Subject: ${subject}`) lines.push(`From: ${from}`) if (to) lines.push(`To: ${to}`) lines.push(`Messages: ${messages.length}`) lines.push('') for (const msg of messages) { const msgFrom = getHeader(msg.payload, 'From') || 'Unknown' const msgDate = getHeader(msg.payload, 'Date') || '' const body = msg.payload ? extractBody(msg.payload) : '' lines.push(`--- ${msgFrom} (${msgDate}) ---`) lines.push(body.trim()) lines.push('') } const firstDate = firstMessage.internalDate ? new Date(Number(firstMessage.internalDate)).toISOString() : undefined const lastDate = lastMessage.internalDate ? new Date(Number(lastMessage.internalDate)).toISOString() : undefined return { content: lines.join('\n').trim(), subject, metadata: { from, to, subject, messageCount: messages.length, labelIds, firstMessageDate: firstDate, lastMessageDate: lastDate, }, } } /** * Fetches a full thread with all its messages. */ async function fetchThread(accessToken: string, threadId: string): Promise { const url = `${GMAIL_API_BASE}/threads/${threadId}?format=FULL` const response = await fetchWithRetry(url, { method: 'GET', headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json', }, }) if (!response.ok) { if (response.status === 404) return null throw new Error(`Failed to fetch thread ${threadId}: ${response.status}`) } return (await response.json()) as GmailThread } /** * Resolves label IDs to human-readable label names using a cache. */ async function resolveLabelNames( accessToken: string, labelIds: string[], syncContext?: Record ): Promise { const cacheKey = '_gmailLabelCache' if (syncContext && !syncContext[cacheKey]) { try { const url = `${GMAIL_API_BASE}/labels` const response = await fetchWithRetry(url, { method: 'GET', headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json', }, }) if (response.ok) { const data = await response.json() const labels = (data.labels || []) as GmailLabel[] const labelMap: Record = {} for (const label of labels) { labelMap[label.id] = label.name } syncContext[cacheKey] = labelMap } } catch { syncContext[cacheKey] = {} } } const cache = (syncContext?.[cacheKey] as Record) ?? {} return labelIds .map((id) => cache[id] || id) .filter((name) => !name.startsWith('CATEGORY_') && name !== 'UNREAD') } /** * Creates a lightweight document stub from a thread list entry. * Uses metadata-based contentHash for change detection without downloading content. */ function threadToStub(thread: { id: string snippet?: string historyId?: string }): ExternalDocument { return { externalId: thread.id, title: thread.snippet || 'Untitled Thread', content: '', contentDeferred: true, mimeType: 'text/plain', sourceUrl: `https://mail.google.com/mail/u/0/#inbox/${thread.id}`, contentHash: `gmail:${thread.id}:${thread.historyId ?? ''}`, metadata: {}, } } export const gmailConnector: ConnectorConfig = { ...gmailConnectorMeta, listDocuments: async ( accessToken: string, sourceConfig: Record, cursor?: string, syncContext?: Record ): Promise => { const searchQuery = buildSearchQuery(sourceConfig) const maxThreads = sourceConfig.maxThreads ? Number(sourceConfig.maxThreads) : DEFAULT_MAX_THREADS const totalFetched = (syncContext?.totalThreadsFetched as number) ?? 0 if (totalFetched >= maxThreads) { return { documents: [], hasMore: false } } const remaining = maxThreads - totalFetched const pageSize = Math.min(THREADS_PER_PAGE, remaining) const queryParams = new URLSearchParams({ maxResults: String(pageSize), }) if (searchQuery) { queryParams.set('q', searchQuery) } if (cursor) { queryParams.set('pageToken', cursor) } const url = `${GMAIL_API_BASE}/threads?${queryParams.toString()}` logger.info('Listing Gmail threads', { query: searchQuery, cursor: cursor ?? 'initial', maxThreads, }) const response = await fetchWithRetry(url, { method: 'GET', headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json', }, }) if (!response.ok) { const errorText = await response.text() logger.error('Failed to list Gmail threads', { status: response.status, error: errorText }) throw new Error(`Failed to list Gmail threads: ${response.status}`) } const data = await response.json() const threads = (data.threads || []) as { id: string; snippet?: string; historyId?: string }[] if (threads.length === 0) { return { documents: [], hasMore: false } } const documents = threads.map(threadToStub) const newTotal = totalFetched + documents.length if (syncContext) syncContext.totalThreadsFetched = newTotal const nextPageToken = data.nextPageToken as string | undefined const hitLimit = newTotal >= maxThreads if (hitLimit && syncContext) syncContext.listingCapped = true return { documents, nextCursor: hitLimit ? undefined : nextPageToken, hasMore: hitLimit ? false : Boolean(nextPageToken), } }, getDocument: async ( accessToken: string, _sourceConfig: Record, externalId: string, syncContext?: Record ): Promise => { try { const thread = await fetchThread(accessToken, externalId) if (!thread) return null const { content, subject, metadata } = formatThread(thread) if (!content.trim()) return null const labelIds = (metadata.labelIds as string[]) || [] const labelNames = await resolveLabelNames(accessToken, labelIds, syncContext) metadata.labels = labelNames return { externalId: thread.id, title: subject, content, contentDeferred: false, mimeType: 'text/plain', sourceUrl: `https://mail.google.com/mail/u/0/#inbox/${thread.id}`, contentHash: `gmail:${thread.id}:${thread.historyId ?? ''}`, metadata, } } catch (error) { logger.warn('Failed to get Gmail thread', { externalId, error: toError(error).message, }) return null } }, validateConfig: async ( accessToken: string, sourceConfig: Record ): Promise<{ valid: boolean; error?: string }> => { const maxThreads = sourceConfig.maxThreads as string | undefined if (maxThreads && (Number.isNaN(Number(maxThreads)) || Number(maxThreads) <= 0)) { return { valid: false, error: 'Max threads must be a positive number' } } try { // Verify Gmail API access by fetching profile const profileUrl = `${GMAIL_API_BASE}/profile` const profileResponse = await fetchWithRetry( profileUrl, { method: 'GET', headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json', }, }, VALIDATE_RETRY_OPTIONS ) if (!profileResponse.ok) { return { valid: false, error: `Failed to access Gmail: ${profileResponse.status}` } } // If labels are specified, verify each one exists const labelNames = parseMultiValue(sourceConfig.label) if (labelNames.length > 0) { const labelsUrl = `${GMAIL_API_BASE}/labels` const labelsResponse = await fetchWithRetry( labelsUrl, { method: 'GET', headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json', }, }, VALIDATE_RETRY_OPTIONS ) if (!labelsResponse.ok) { return { valid: false, error: 'Failed to fetch labels' } } const labelsData = await labelsResponse.json() const labels = (labelsData.labels || []) as GmailLabel[] const labelNameSet = new Set(labels.map((l) => l.name.toLowerCase())) const missing = labelNames.filter((name) => !labelNameSet.has(name.toLowerCase())) if (missing.length > 0) { return { valid: false, error: `Label(s) not found: ${missing.join(', ')}. Available labels: ${labels .filter( (l) => l.type !== 'system' || ['INBOX', 'IMPORTANT', 'STARRED', 'SENT', 'DRAFT'].includes(l.id) ) .map((l) => l.name) .slice(0, 15) .join(', ')}`, } } } // If a custom query is specified, verify it's valid by doing a dry-run const query = sourceConfig.query as string | undefined if (query?.trim()) { const searchQuery = buildSearchQuery(sourceConfig) const testUrl = `${GMAIL_API_BASE}/threads?q=${encodeURIComponent(searchQuery)}&maxResults=1` const testResponse = await fetchWithRetry( testUrl, { method: 'GET', headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json', }, }, VALIDATE_RETRY_OPTIONS ) if (!testResponse.ok) { return { valid: false, error: 'Invalid search query. Check Gmail search syntax.' } } } return { valid: true } } catch (error) { const message = getErrorMessage(error, 'Failed to validate configuration') return { valid: false, error: message } } }, mapTags: (metadata: Record): Record => { const result: Record = {} if (typeof metadata.from === 'string') { result.from = metadata.from } const labels = joinTagArray(metadata.labels) if (labels) { result.labels = labels } if (typeof metadata.messageCount === 'number') { result.messageCount = metadata.messageCount } const lastMessageDate = parseTagDate(metadata.lastMessageDate) if (lastMessageDate) { result.lastMessageDate = lastMessageDate } return result }, }