import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/documents/utils' import { salesforceConnectorMeta } from '@/connectors/salesforce/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { htmlToPlainText, parseTagDate } from '@/connectors/utils' const logger = createLogger('SalesforceConnector') /** * Salesforce serves the userinfo endpoint at the org's authentication host. * Tokens issued at test.salesforce.com (sandbox) are rejected at login.salesforce.com, * so we try each host in order and cache the working one in syncContext. */ const USERINFO_HOSTS = ['https://login.salesforce.com', 'https://test.salesforce.com'] as const const USERINFO_PATH = '/services/oauth2/userinfo' const API_VERSION = 'v62.0' const PAGE_SIZE = 200 /** SOQL field lists per object type. */ const OBJECT_FIELDS: Record = { KnowledgeArticleVersion: [ 'Id', 'Title', 'Summary', 'LastModifiedDate', 'ArticleNumber', 'PublishStatus', ], Case: ['Id', 'Subject', 'Description', 'Status', 'LastModifiedDate', 'CaseNumber'], Account: ['Id', 'Name', 'Description', 'Industry', 'LastModifiedDate'], Opportunity: [ 'Id', 'Name', 'Description', 'StageName', 'Amount', 'LastModifiedDate', 'CloseDate', ], } as const /** SOQL WHERE clause additions per object type. */ const OBJECT_WHERE: Record = { KnowledgeArticleVersion: " WHERE PublishStatus='Online' AND IsLatestVersion=true AND Language='en_US'", } as const /** * Result of a userinfo lookup: either the parsed payload + the auth host that * served it, or a structured failure describing the last response we saw. */ type UserinfoResult = | { ok: true; data: Record; host: string } | { ok: false; status: number | undefined; errorText: string } /** * Fetches the Salesforce userinfo payload, trying each candidate auth host in * order. Sandbox-issued tokens are rejected at login.salesforce.com with 401/403, * so on those statuses we fall through to test.salesforce.com. The working host * is cached in syncContext under `_salesforceInstanceUrl` so subsequent calls in * the same sync run skip the fallback dance. */ async function fetchUserinfo( accessToken: string, retryOptions?: Parameters[2], syncContext?: Record ): Promise { const cachedHost = typeof syncContext?._salesforceInstanceUrl === 'string' ? (syncContext._salesforceInstanceUrl as string) : undefined const orderedHosts = cachedHost ? [cachedHost, ...USERINFO_HOSTS.filter((h) => h !== cachedHost)] : [...USERINFO_HOSTS] let lastStatus: number | undefined let lastErrorText = '' for (const host of orderedHosts) { const response = await fetchWithRetry( `${host}${USERINFO_PATH}`, { method: 'GET', headers: { Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, }, retryOptions ) if (response.ok) { const data = (await response.json()) as Record if (syncContext) { syncContext._salesforceInstanceUrl = host } return { ok: true, data, host } } lastStatus = response.status lastErrorText = await response.text() // Only fall through to the next host on auth-shaped failures; surface // other errors (e.g. 5xx) immediately so we don't mask real problems. if (response.status !== 401 && response.status !== 403) { break } } return { ok: false, status: lastStatus, errorText: lastErrorText } } /** * Resolves the Salesforce instance REST URL from the userinfo endpoint. * Caches the result in syncContext to avoid repeated calls. */ async function resolveInstanceUrl( accessToken: string, syncContext?: Record ): Promise { if (syncContext?.instanceUrl) { return syncContext.instanceUrl as string } const result = await fetchUserinfo(accessToken, undefined, syncContext) if (!result.ok) { throw new Error( `Failed to resolve Salesforce instance URL: ${result.status ?? 'unknown'} - ${result.errorText}` ) } const urls = result.data.urls as Record | undefined let restUrl = urls?.rest if (!restUrl) { throw new Error('Salesforce userinfo response did not include a REST URL') } restUrl = restUrl.replace('{version}', API_VERSION) if (syncContext) { syncContext.instanceUrl = restUrl } return restUrl } /** * Builds the document title for a Salesforce record based on its object type. */ function buildRecordTitle(objectType: string, record: Record): string { switch (objectType) { case 'KnowledgeArticleVersion': return (record.Title as string) || 'Untitled Article' case 'Case': return (record.Subject as string) || 'Untitled Case' case 'Account': return (record.Name as string) || 'Unnamed Account' case 'Opportunity': return (record.Name as string) || 'Unnamed Opportunity' default: return `Record ${(record.Id as string) || 'Unknown'}` } } /** Fields that may contain HTML content and should be stripped to plain text. */ const HTML_FIELDS = new Set(['Description', 'Summary']) /** * Builds plain-text content from a Salesforce record for indexing. */ function buildRecordContent(objectType: string, record: Record): string { const parts: string[] = [] const title = buildRecordTitle(objectType, record) parts.push(title) const fields = OBJECT_FIELDS[objectType] || [] for (const field of fields) { if (field === 'Id') continue const value = record[field] if (value != null && value !== '') { const label = field.replace(/([A-Z])/g, ' $1').trim() const text = HTML_FIELDS.has(field) && typeof value === 'string' ? htmlToPlainText(value) : String(value) parts.push(`${label}: ${text}`) } } return parts.join('\n').trim() } /** * Returns the record number field value based on object type. */ function getRecordNumber(objectType: string, record: Record): string | undefined { switch (objectType) { case 'KnowledgeArticleVersion': return (record.ArticleNumber as string) || undefined case 'Case': return (record.CaseNumber as string) || undefined default: return undefined } } /** * Returns the status/stage field value based on object type. */ function getRecordStatus(objectType: string, record: Record): string | undefined { switch (objectType) { case 'Case': return (record.Status as string) || undefined case 'Opportunity': return (record.StageName as string) || undefined default: return undefined } } /** * Creates a lightweight stub for a Salesforce record with metadata-based hash. * Content is deferred and fetched later via getDocument only for new/changed docs. */ function recordToStub( record: Record, objectType: string, instanceUrl: string ): ExternalDocument { const id = record.Id as string const title = buildRecordTitle(objectType, record) const lastModified = (record.LastModifiedDate as string) || '' const baseUrl = instanceUrl.replace(`/services/data/${API_VERSION}/`, '') return { externalId: id, title, content: '', contentDeferred: true, mimeType: 'text/plain', sourceUrl: `${baseUrl}/${id}`, contentHash: `salesforce:${id}:${lastModified}`, metadata: { objectType, lastModified: lastModified || undefined, recordNumber: getRecordNumber(objectType, record), status: getRecordStatus(objectType, record), }, } } /** * Builds a full ExternalDocument with content from a Salesforce record. */ function recordToDocument( record: Record, objectType: string, instanceUrl: string ): ExternalDocument { const stub = recordToStub(record, objectType, instanceUrl) return { ...stub, content: buildRecordContent(objectType, record), contentDeferred: false, } } export const salesforceConnector: ConnectorConfig = { ...salesforceConnectorMeta, listDocuments: async ( accessToken: string, sourceConfig: Record, cursor?: string, syncContext?: Record ): Promise => { const objectType = sourceConfig.objectType as string const maxRecords = sourceConfig.maxRecords ? Number(sourceConfig.maxRecords) : 0 const fields = OBJECT_FIELDS[objectType] if (!fields) { throw new Error(`Unsupported Salesforce object type: ${objectType}`) } const instanceUrl = await resolveInstanceUrl(accessToken, syncContext) let url: string if (cursor) { const baseUrl = instanceUrl.replace(`/services/data/${API_VERSION}/`, '') url = `${baseUrl}${cursor}` } else { const whereClause = OBJECT_WHERE[objectType] || '' const soql = `SELECT ${fields.join(',')} FROM ${objectType}${whereClause} ORDER BY LastModifiedDate DESC LIMIT ${PAGE_SIZE}` url = `${instanceUrl}query?q=${encodeURIComponent(soql)}` } logger.info(`Listing Salesforce ${objectType}`, { cursor: cursor || 'initial' }) const response = await fetchWithRetry(url, { method: 'GET', headers: { Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, }) if (!response.ok) { const errorText = await response.text() logger.error(`Failed to query Salesforce ${objectType}`, { status: response.status, error: errorText, }) throw new Error(`Failed to query Salesforce ${objectType}: ${response.status}`) } const data = await response.json() const records = (data.records || []) as Record[] const nextRecordsUrl = data.nextRecordsUrl as string | undefined const documents: ExternalDocument[] = records.map((record) => recordToStub(record, objectType, instanceUrl) ) const previouslyFetched = (syncContext?.totalDocsFetched as number) ?? 0 if (maxRecords > 0) { const remaining = maxRecords - previouslyFetched if (documents.length > remaining) { documents.splice(remaining) } } const totalFetched = previouslyFetched + documents.length if (syncContext) { syncContext.totalDocsFetched = totalFetched } const hasMore = Boolean(nextRecordsUrl) && (maxRecords <= 0 || totalFetched < maxRecords) return { documents, nextCursor: hasMore ? nextRecordsUrl : undefined, hasMore, } }, getDocument: async ( accessToken: string, sourceConfig: Record, externalId: string, syncContext?: Record ): Promise => { const objectType = sourceConfig.objectType as string const fields = OBJECT_FIELDS[objectType] if (!fields) { throw new Error(`Unsupported Salesforce object type: ${objectType}`) } let instanceUrl = syncContext?.instanceUrl as string | undefined if (!instanceUrl) { instanceUrl = await resolveInstanceUrl(accessToken, syncContext) } const url = `${instanceUrl}sobjects/${objectType}/${encodeURIComponent(externalId)}?fields=${fields.join(',')}` const response = await fetchWithRetry(url, { method: 'GET', headers: { Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, }) if (!response.ok) { if (response.status === 404) return null throw new Error(`Failed to get Salesforce ${objectType} record: ${response.status}`) } const record = await response.json() if ( objectType === 'KnowledgeArticleVersion' && (record as Record).PublishStatus !== 'Online' ) { return null } return recordToDocument(record, objectType, instanceUrl) }, validateConfig: async ( accessToken: string, sourceConfig: Record ): Promise<{ valid: boolean; error?: string }> => { const objectType = sourceConfig.objectType as string if (!objectType) { return { valid: false, error: 'Object type is required' } } if (!OBJECT_FIELDS[objectType]) { return { valid: false, error: `Unsupported object type: ${objectType}` } } const maxRecords = sourceConfig.maxRecords as string | undefined if (maxRecords && (Number.isNaN(Number(maxRecords)) || Number(maxRecords) <= 0)) { return { valid: false, error: 'Max records must be a positive number' } } try { const userinfoResult = await fetchUserinfo(accessToken, VALIDATE_RETRY_OPTIONS) if (!userinfoResult.ok) { return { valid: false, error: `Failed to authenticate with Salesforce: ${userinfoResult.status ?? 'unknown'} - ${userinfoResult.errorText}`, } } const urls = userinfoResult.data.urls as Record | undefined let restUrl = urls?.rest if (!restUrl) { return { valid: false, error: 'Could not resolve Salesforce instance URL' } } restUrl = restUrl.replace('{version}', API_VERSION) const soql = `SELECT Id FROM ${objectType} LIMIT 1` const queryUrl = `${restUrl}query?q=${encodeURIComponent(soql)}` const queryResponse = await fetchWithRetry( queryUrl, { method: 'GET', headers: { Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, }, VALIDATE_RETRY_OPTIONS ) if (!queryResponse.ok) { const errorText = await queryResponse.text() return { valid: false, error: `Failed to access Salesforce ${objectType}: ${queryResponse.status} - ${errorText}`, } } return { valid: true } } catch (error) { return { valid: false, error: toError(error).message || 'Failed to validate configuration' } } }, mapTags: (metadata: Record): Record => { const result: Record = {} if (typeof metadata.objectType === 'string') result.objectType = metadata.objectType const lastModified = parseTagDate(metadata.lastModified) if (lastModified) result.lastModified = lastModified if (typeof metadata.recordNumber === 'string') result.recordNumber = metadata.recordNumber if (typeof metadata.status === 'string') result.status = metadata.status return result }, }