import { createLogger } from '@sim/logger' import { getErrorMessage, toError } from '@sim/utils/errors' import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/documents/utils' import { fathomConnectorMeta } from '@/connectors/fathom/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { parseTagDate } from '@/connectors/utils' const logger = createLogger('FathomConnector') const FATHOM_API_BASE = 'https://api.fathom.ai/external/v1' const MS_PER_DAY = 24 * 60 * 60 * 1000 /** * Days subtracted from `lastSyncAt` when computing the incremental `created_after` * window. Fathom's list endpoint only filters by creation time (no update-based * filter), so a meeting whose transcript was not yet ready on the sync that first * saw it would otherwise never be re-listed. The overlap keeps recently-created * meetings in the window long enough for late transcripts to be retried — the sync * engine re-attempts meetings whose `getDocument` previously returned null, since * those are never persisted. Matches the Gong connector's overlap approach. */ const INCREMENTAL_OVERLAP_DAYS = 14 /** * Fathom authenticates external API requests with the `X-Api-Key` header. * (The API also accepts `Authorization: Bearer` for OAuth-connected apps, but * the api-key flow this connector uses requires `X-Api-Key`.) */ function buildHeaders(apiKey: string): Record { return { 'X-Api-Key': apiKey, 'Content-Type': 'application/json', } } /** * A meeting object as returned by `GET /meetings`. Only the fields this * connector reads are typed; the API returns additional fields. */ interface FathomMeeting { recording_id?: number title?: string meeting_title?: string | null url?: string share_url?: string created_at?: string scheduled_start_time?: string | null scheduled_end_time?: string | null recording_start_time?: string | null recording_end_time?: string | null transcript_language?: string calendar_invitees_domains_type?: 'only_internal' | 'one_or_more_external' | null recorded_by?: { name?: string email?: string email_domain?: string team?: string | null } | null } interface FathomMeetingsListResponse { items?: FathomMeeting[] next_cursor?: string | null } /** * A single transcript entry as returned by `GET /recordings/{id}/transcript`. */ interface FathomTranscriptEntry { speaker?: { display_name?: string matched_calendar_invitee_email?: string | null } text?: string timestamp?: string } interface FathomTranscriptResponse { transcript?: FathomTranscriptEntry[] } interface FathomSummary { template_name?: string | null markdown_formatted?: string | null } interface FathomSummaryResponse { summary?: FathomSummary | null } /** * Header fields cached per recording during `listDocuments` so `getDocument` * can render an identical document header. Fathom exposes no single-meeting * GET and no `recording_ids` filter, so this metadata cannot be refetched once * listing has moved past the page that contained it — it is carried forward in * the shared `syncContext` instead. */ interface FathomMeetingHeader { title: string meetingDate?: string durationSeconds?: number recordedByEmail?: string recordedByName?: string team?: string sourceUrl?: string contentHash: string metadata: FathomMeetingMetadata } /** * Metadata describing a Fathom meeting, attached to the listing stub. The sync * engine merges this onto the hydrated document, so `getDocument` never needs * to reproduce it. */ interface FathomMeetingMetadata { recordingId?: string recordedByEmail?: string recordedByName?: string team?: string meetingType?: string meetingDate?: string durationSeconds?: number transcriptLanguage?: string title?: string } /** * Maps Fathom's `calendar_invitees_domains_type` to a human-readable meeting * type. This is the only meeting-type signal the API exposes: `only_internal` * means every invitee shares the recorder's domain; `one_or_more_external` * means at least one external attendee (customer-facing). */ function resolveMeetingType(meeting: FathomMeeting): string | undefined { switch (meeting.calendar_invitees_domains_type) { case 'only_internal': return 'internal' case 'one_or_more_external': return 'external' default: return undefined } } /** * Computes the meeting duration in whole seconds from the recording window, * or undefined when either bound is missing or unparseable. */ function computeDurationSeconds(meeting: FathomMeeting): number | undefined { const start = meeting.recording_start_time ?? meeting.scheduled_start_time ?? undefined const end = meeting.recording_end_time ?? meeting.scheduled_end_time ?? undefined if (!start || !end) return undefined const startMs = new Date(start).getTime() const endMs = new Date(end).getTime() if (Number.isNaN(startMs) || Number.isNaN(endMs) || endMs < startMs) return undefined return Math.round((endMs - startMs) / 1000) } /** * Returns the best title for a meeting, falling back through the title fields. */ function resolveTitle(meeting: FathomMeeting): string { const title = meeting.title?.trim() || meeting.meeting_title?.trim() return title || 'Untitled Fathom Meeting' } /** * Extracts the connector metadata bag attached to a listing stub. */ function buildMetadata(meeting: FathomMeeting): FathomMeetingMetadata { return { recordingId: meeting.recording_id != null ? String(meeting.recording_id) : undefined, recordedByEmail: meeting.recorded_by?.email, recordedByName: meeting.recorded_by?.name, team: meeting.recorded_by?.team ?? undefined, meetingType: resolveMeetingType(meeting), meetingDate: meeting.recording_start_time ?? meeting.created_at ?? undefined, durationSeconds: computeDurationSeconds(meeting), transcriptLanguage: meeting.transcript_language, title: resolveTitle(meeting), } } /** * Extracts the lightweight header fields cached for `getDocument`. */ function buildHeader(meeting: FathomMeeting): FathomMeetingHeader { return { title: resolveTitle(meeting), meetingDate: meeting.recording_start_time ?? meeting.created_at ?? undefined, durationSeconds: computeDurationSeconds(meeting), recordedByEmail: meeting.recorded_by?.email, recordedByName: meeting.recorded_by?.name, team: meeting.recorded_by?.team ?? undefined, sourceUrl: buildSourceUrl(meeting), contentHash: buildContentHash(meeting), metadata: buildMetadata(meeting), } } /** * Builds a metadata-based content hash. Fathom recordings are immutable once * processed, so the recording id plus its end/creation timestamps fully identify * a version. The same value is cached in the header and returned by `getDocument`, * so the stub and hydrated document hash identically. */ function buildContentHash(meeting: FathomMeeting): string { return `fathom:${meeting.recording_id ?? ''}:${meeting.recording_end_time ?? ''}:${meeting.created_at ?? ''}` } function buildSourceUrl(meeting: FathomMeeting): string | undefined { return meeting.share_url || meeting.url || undefined } /** * Reads the cached header for a recording out of the shared sync context. */ function readCachedHeader( syncContext: Record | undefined, recordingId: string ): FathomMeetingHeader | undefined { const cache = syncContext?.meetingHeaders as Record | undefined return cache?.[recordingId] } /** * Stores the header for a recording in the shared sync context. */ function cacheHeader( syncContext: Record | undefined, recordingId: string, header: FathomMeetingHeader ): void { if (!syncContext) return const cache = (syncContext.meetingHeaders as Record | undefined) ?? {} cache[recordingId] = header syncContext.meetingHeaders = cache } /** * Formats the meeting header, optional summary, and transcript into a single * plain-text document with one `Speaker: text` line per transcript entry. */ function formatMeetingContent( header: FathomMeetingHeader | undefined, transcript: FathomTranscriptEntry[], summary: FathomSummary | null ): string { const parts: string[] = [] parts.push(`Meeting: ${header?.title ?? 'Untitled Fathom Meeting'}`) if (header?.meetingDate) parts.push(`Date: ${header.meetingDate}`) if (header?.durationSeconds != null) { parts.push(`Duration: ${Math.round(header.durationSeconds / 60)} minutes`) } if (header?.recordedByEmail) { parts.push(`Recorded by: ${header.recordedByName ?? header.recordedByEmail}`) } if (header?.team) parts.push(`Team: ${header.team}`) if (summary?.markdown_formatted?.trim()) { parts.push('') parts.push('--- Summary ---') parts.push(summary.markdown_formatted.trim()) } if (transcript.length > 0) { parts.push('') parts.push('--- Transcript ---') for (const entry of transcript) { const speaker = entry.speaker?.display_name?.trim() || 'Unknown' const text = entry.text?.trim() if (text) parts.push(`${speaker}: ${text}`) } } return parts.join('\n') } /** * Converts a listing meeting into a deferred stub. Content is fetched lazily * via `getDocument` only for new or changed meetings. */ function meetingToStub(meeting: FathomMeeting): ExternalDocument { const metadata = buildMetadata(meeting) return { externalId: String(meeting.recording_id), title: resolveTitle(meeting), content: '', contentDeferred: true, mimeType: 'text/plain', sourceUrl: buildSourceUrl(meeting), contentHash: buildContentHash(meeting), metadata: { ...metadata }, } } export const fathomConnector: ConnectorConfig = { ...fathomConnectorMeta, listDocuments: async ( accessToken: string, sourceConfig: Record, cursor?: string, syncContext?: Record, lastSyncAt?: Date ): Promise => { const recordedBy = (sourceConfig.recordedBy as string | undefined)?.trim() const teams = (sourceConfig.teams as string | undefined)?.trim() const meetingType = (sourceConfig.meetingType as string | undefined)?.trim() const inviteeDomain = (sourceConfig.inviteeDomains as string | undefined)?.trim() const maxMeetings = sourceConfig.maxMeetings ? Number(sourceConfig.maxMeetings) : 0 const url = new URL(`${FATHOM_API_BASE}/meetings`) if (recordedBy) url.searchParams.append('recorded_by[]', recordedBy) if (teams) url.searchParams.append('teams[]', teams) if (meetingType && meetingType !== 'all') { url.searchParams.append('calendar_invitees_domains_type', meetingType) } if (inviteeDomain) url.searchParams.append('calendar_invitees_domains[]', inviteeDomain) if (cursor) url.searchParams.append('cursor', cursor) if (lastSyncAt) { const createdAfter = new Date(lastSyncAt.getTime() - INCREMENTAL_OVERLAP_DAYS * MS_PER_DAY) url.searchParams.append('created_after', createdAfter.toISOString()) } logger.info('Listing Fathom meetings', { hasCursor: Boolean(cursor), recordedBy, teams, meetingType, inviteeDomain, incremental: Boolean(lastSyncAt), }) const response = await fetchWithRetry(url.toString(), { method: 'GET', headers: buildHeaders(accessToken), }) if (!response.ok) { const errorText = await response.text().catch(() => '') logger.error('Failed to list Fathom meetings', { status: response.status, error: errorText.slice(0, 500), }) throw new Error(`Failed to list Fathom meetings: ${response.status}`) } const data = (await response.json()) as FathomMeetingsListResponse const meetings = data.items ?? [] const nextCursor = data.next_cursor?.trim() || undefined const allDocuments: ExternalDocument[] = [] for (const meeting of meetings) { if (meeting.recording_id == null) continue const externalId = String(meeting.recording_id) cacheHeader(syncContext, externalId, buildHeader(meeting)) allDocuments.push(meetingToStub(meeting)) } const prevFetched = (syncContext?.totalDocsFetched as number) ?? 0 let documents = allDocuments if (maxMeetings > 0) { const remaining = Math.max(0, maxMeetings - prevFetched) if (allDocuments.length > remaining) { documents = allDocuments.slice(0, remaining) } } const totalFetched = prevFetched + documents.length if (syncContext) syncContext.totalDocsFetched = totalFetched const hitLimit = maxMeetings > 0 && totalFetched >= maxMeetings if (hitLimit && syncContext) syncContext.listingCapped = true const hasMore = !hitLimit && Boolean(nextCursor) return { documents, nextCursor: hasMore ? nextCursor : undefined, hasMore, } }, getDocument: async ( accessToken: string, _sourceConfig: Record, externalId: string, syncContext?: Record ): Promise => { try { if (!externalId) return null const transcriptUrl = `${FATHOM_API_BASE}/recordings/${encodeURIComponent(externalId)}/transcript` const transcriptResponse = await fetchWithRetry(transcriptUrl, { method: 'GET', headers: buildHeaders(accessToken), }) if (!transcriptResponse.ok) { if (transcriptResponse.status === 404) return null throw new Error(`Failed to fetch Fathom transcript: ${transcriptResponse.status}`) } const transcriptData = (await transcriptResponse.json()) as FathomTranscriptResponse const transcript = transcriptData.transcript ?? [] let summary: FathomSummary | null = null try { const summaryUrl = `${FATHOM_API_BASE}/recordings/${encodeURIComponent(externalId)}/summary` const summaryResponse = await fetchWithRetry(summaryUrl, { method: 'GET', headers: buildHeaders(accessToken), }) if (summaryResponse.ok) { const summaryData = (await summaryResponse.json()) as FathomSummaryResponse summary = summaryData.summary ?? null } } catch (summaryError) { logger.warn('Failed to fetch Fathom summary', { externalId, error: toError(summaryError).message, }) } const hasTranscript = transcript.some((entry) => entry.text?.trim()) const hasSummary = Boolean(summary?.markdown_formatted?.trim()) if (!hasTranscript && !hasSummary) { logger.info('No transcript or summary yet for Fathom meeting', { externalId }) return null } const header = readCachedHeader(syncContext, externalId) if (!header) { logger.warn( 'No cached header for Fathom meeting; skipping to avoid an un-refreshable record', { externalId, } ) return null } const content = formatMeetingContent(header, transcript, summary).trim() if (!content) return null return { externalId, title: header.title, content, contentDeferred: false, mimeType: 'text/plain', sourceUrl: header.sourceUrl, contentHash: header.contentHash, metadata: { ...header.metadata }, } } catch (error) { logger.warn('Failed to get Fathom meeting', { externalId, error: toError(error).message, }) return null } }, validateConfig: async ( accessToken: string, sourceConfig: Record ): Promise<{ valid: boolean; error?: string }> => { const maxMeetings = sourceConfig.maxMeetings as string | undefined if (maxMeetings && (Number.isNaN(Number(maxMeetings)) || Number(maxMeetings) < 0)) { return { valid: false, error: 'Max meetings must be a non-negative number' } } try { const response = await fetchWithRetry( `${FATHOM_API_BASE}/meetings`, { method: 'GET', headers: buildHeaders(accessToken), }, VALIDATE_RETRY_OPTIONS ) if (!response.ok) { const errorText = await response.text().catch(() => '') return { valid: false, error: `Fathom access failed: ${response.status}${errorText ? ` — ${errorText.slice(0, 200)}` : ''}`, } } 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.title === 'string' && metadata.title.trim()) { result.title = metadata.title } if (typeof metadata.recordedByEmail === 'string' && metadata.recordedByEmail.trim()) { result.recordedByEmail = metadata.recordedByEmail } if (typeof metadata.recordedByName === 'string' && metadata.recordedByName.trim()) { result.recordedByName = metadata.recordedByName } if (typeof metadata.team === 'string' && metadata.team.trim()) { result.team = metadata.team } if (typeof metadata.meetingType === 'string' && metadata.meetingType.trim()) { result.meetingType = metadata.meetingType } if (typeof metadata.transcriptLanguage === 'string' && metadata.transcriptLanguage.trim()) { result.transcriptLanguage = metadata.transcriptLanguage } if (metadata.durationSeconds != null) { const num = Number(metadata.durationSeconds) if (!Number.isNaN(num)) result.durationSeconds = num } const meetingDate = parseTagDate(metadata.meetingDate) if (meetingDate) result.meetingDate = meetingDate return result }, }