import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/documents/utils' import { hubspotConnectorMeta } from '@/connectors/hubspot/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { parseTagDate } from '@/connectors/utils' const logger = createLogger('HubSpotConnector') const BASE_URL = 'https://api.hubapi.com' const PAGE_SIZE = 100 /** Properties to fetch per object type. */ const OBJECT_PROPERTIES: Record = { contacts: [ 'firstname', 'lastname', 'email', 'phone', 'company', 'jobtitle', 'lifecyclestage', 'hs_lead_status', 'hubspot_owner_id', 'createdate', 'lastmodifieddate', ], companies: [ 'name', 'domain', 'industry', 'description', 'phone', 'city', 'state', 'country', 'numberofemployees', 'annualrevenue', 'hubspot_owner_id', 'createdate', 'hs_lastmodifieddate', ], deals: [ 'dealname', 'amount', 'dealstage', 'pipeline', 'closedate', 'hubspot_owner_id', 'createdate', 'hs_lastmodifieddate', ], tickets: [ 'subject', 'content', 'hs_pipeline', 'hs_pipeline_stage', 'hs_ticket_priority', 'hubspot_owner_id', 'createdate', 'hs_lastmodifieddate', ], } as const /** * Fetches the HubSpot portal ID for the authenticated account. * Caches the result in syncContext to avoid repeated calls. */ async function getPortalId( accessToken: string, syncContext?: Record ): Promise { if (syncContext?.portalId) { return syncContext.portalId as string } const response = await fetchWithRetry(`${BASE_URL}/account-info/v3/details`, { method: 'GET', headers: { Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, }) if (!response.ok) { const errorText = await response.text() throw new Error(`Failed to fetch HubSpot portal ID: ${response.status} - ${errorText}`) } const data = await response.json() const portalId = String(data.portalId) if (syncContext) { syncContext.portalId = portalId } return portalId } /** * Builds the document title for a CRM record based on its object type. */ function buildRecordTitle(objectType: string, properties: Record): string { switch (objectType) { case 'contacts': { const first = properties.firstname || '' const last = properties.lastname || '' const name = `${first} ${last}`.trim() return name || properties.email || 'Unnamed Contact' } case 'companies': return properties.name || properties.domain || 'Unnamed Company' case 'deals': return properties.dealname || 'Unnamed Deal' case 'tickets': return properties.subject || 'Unnamed Ticket' default: return `Record ${properties.hs_object_id || 'Unknown'}` } } /** * Builds a plain-text representation of a CRM record's properties for indexing. */ function buildRecordContent(objectType: string, properties: Record): string { const parts: string[] = [] const title = buildRecordTitle(objectType, properties) parts.push(title) for (const [key, value] of Object.entries(properties)) { if (value && key !== 'hs_object_id') { const label = key.replace(/_/g, ' ').replace(/\b\w/g, (c) => c.toUpperCase()) parts.push(`${label}: ${value}`) } } return parts.join('\n').trim() } /** * Converts a HubSpot CRM record to an ExternalDocument. */ function recordToDocument( record: Record, objectType: string, portalId: string ): ExternalDocument { const id = record.id as string const properties = (record.properties || {}) as Record const content = buildRecordContent(objectType, properties) const title = buildRecordTitle(objectType, properties) const lastModified = properties.lastmodifieddate || properties.hs_lastmodifieddate || properties.createdate return { externalId: id, title, content, mimeType: 'text/plain', sourceUrl: `https://app.hubspot.com/contacts/${portalId}/record/${objectType}/${id}`, contentHash: `hubspot:${id}:${lastModified ?? ''}`, metadata: { objectType, owner: properties.hubspot_owner_id || undefined, lastModified: lastModified || undefined, pipeline: properties.pipeline || properties.hs_pipeline || undefined, }, } } export const hubspotConnector: ConnectorConfig = { ...hubspotConnectorMeta, 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 properties = OBJECT_PROPERTIES[objectType] || [] const portalId = await getPortalId(accessToken, syncContext) const sortProperty = objectType === 'contacts' ? 'lastmodifieddate' : 'hs_lastmodifieddate' const searchBody: Record = { properties, sorts: [{ propertyName: sortProperty, direction: 'DESCENDING' }], limit: PAGE_SIZE, } if (cursor) { searchBody.after = cursor } logger.info(`Listing HubSpot ${objectType}`, { cursor }) const response = await fetchWithRetry(`${BASE_URL}/crm/v3/objects/${objectType}/search`, { method: 'POST', headers: { 'Content-Type': 'application/json', Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, body: JSON.stringify(searchBody), }) if (!response.ok) { const errorText = await response.text() logger.error(`Failed to list HubSpot ${objectType}`, { status: response.status, error: errorText, }) throw new Error(`Failed to list HubSpot ${objectType}: ${response.status}`) } const data = await response.json() const results = (data.results || []) as Record[] const paging = data.paging as { next?: { after?: string } } | undefined const nextCursor = paging?.next?.after const documents: ExternalDocument[] = results.map((record) => recordToDocument(record, objectType, portalId) ) 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(nextCursor) && (maxRecords <= 0 || totalFetched < maxRecords) return { documents, nextCursor: hasMore ? nextCursor : undefined, hasMore, } }, getDocument: async ( accessToken: string, sourceConfig: Record, externalId: string, syncContext?: Record ): Promise => { const objectType = sourceConfig.objectType as string const properties = OBJECT_PROPERTIES[objectType] || [] let portalId = syncContext?.portalId as string | undefined if (!portalId) { portalId = await getPortalId(accessToken) if (syncContext) syncContext.portalId = portalId } const params = new URLSearchParams() for (const prop of properties) { params.append('properties', prop) } const url = `${BASE_URL}/crm/v3/objects/${objectType}/${externalId}?${params.toString()}` 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 HubSpot ${objectType} record: ${response.status}`) } const record = await response.json() return recordToDocument(record as Record, objectType, portalId) }, 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_PROPERTIES[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 response = await fetchWithRetry( `${BASE_URL}/crm/v3/objects/${objectType}?limit=1`, { method: 'GET', headers: { Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, }, VALIDATE_RETRY_OPTIONS ) if (!response.ok) { const errorText = await response.text() return { valid: false, error: `Failed to access HubSpot ${objectType}: ${response.status} - ${errorText}`, } } 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.objectType === 'string') result.objectType = metadata.objectType if (typeof metadata.owner === 'string') result.owner = metadata.owner const lastModified = parseTagDate(metadata.lastModified) if (lastModified) result.lastModified = lastModified if (typeof metadata.pipeline === 'string') result.pipeline = metadata.pipeline return result }, }