import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/documents/utils' import { confluenceConnectorMeta } from '@/connectors/confluence/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { htmlToPlainText, joinTagArray, parseMultiValue, parseTagDate } from '@/connectors/utils' import { getConfluenceCloudId, normalizeConfluenceDomainHost } from '@/tools/confluence/utils' const logger = createLogger('ConfluenceConnector') /** * Escapes a value for use inside CQL double-quoted strings. */ export function escapeCql(value: string): string { return value.replace(/\\/g, '\\\\').replace(/"/g, '\\"') } /** * Builds a CQL clause restricting content to the given space keys. * Single key uses `space = "X"`; multiple keys use `space in ("X","Y")`. */ function buildSpaceClause(spaceKeys: string[]): string { if (spaceKeys.length === 1) { return `space="${escapeCql(spaceKeys[0])}"` } const list = spaceKeys.map((k) => `"${escapeCql(k)}"`).join(',') return `space in (${list})` } /** * Fetches labels for a batch of page IDs using the v2 labels endpoint. */ const LABEL_FETCH_CONCURRENCY = 5 async function fetchLabelsForPages( cloudId: string, accessToken: string, pageIds: string[] ): Promise> { const labelsByPageId = new Map() for (let i = 0; i < pageIds.length; i += LABEL_FETCH_CONCURRENCY) { const batch = pageIds.slice(i, i + LABEL_FETCH_CONCURRENCY) const results = await Promise.all( batch.map(async (pageId) => { try { let data: Record | null = null for (const contentType of ['pages', 'blogposts']) { const url = `https://api.atlassian.com/ex/confluence/${cloudId}/wiki/api/v2/${contentType}/${pageId}/labels` const response = await fetchWithRetry(url, { method: 'GET', headers: { Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, }) if (response.ok) { data = await response.json() break } if (response.status !== 404) { logger.warn(`Failed to fetch labels for ${contentType} ${pageId}`, { status: response.status, }) } } if (!data) { return { pageId, labels: [] as string[] } } const labels = ((data.results as Record[]) || []).map( (label) => label.name as string ) return { pageId, labels } } catch (error) { logger.warn(`Error fetching labels for page ${pageId}`, { error: toError(error).message, }) return { pageId, labels: [] as string[] } } }) ) for (const { pageId, labels } of results) { labelsByPageId.set(pageId, labels) } } return labelsByPageId } /** * Produces a canonical metadata stub with a deterministic contentHash that * does not depend on which API surface (v1 CQL or v2) returned the page. */ function pageToStub( page: Record, options: { spaceId?: unknown labels?: string[] sourceUrl?: string } = {} ): ExternalDocument { const version = page.version as Record | undefined const versionNumber = version?.number as number | undefined const lastModified = (version?.createdAt ?? version?.when ?? '') as string const versionKey = versionNumber ?? lastModified return { externalId: String(page.id), title: (page.title as string) || 'Untitled', content: '', contentDeferred: true, mimeType: 'text/plain', sourceUrl: options.sourceUrl, contentHash: `confluence:${page.id}:${versionKey}`, metadata: { spaceId: options.spaceId, status: page.status, version: versionNumber, labels: options.labels ?? [], lastModified, }, } } /** * Converts a v1 CQL search result item to a lightweight metadata stub. */ function cqlResultToStub(item: Record, domain: string): ExternalDocument { const links = item._links as Record | undefined const metadata = item.metadata as Record | undefined const labelsWrapper = metadata?.labels as Record | undefined const labelResults = (labelsWrapper?.results || []) as Record[] const labels = labelResults.map((l) => l.name as string) return pageToStub(item, { spaceId: (item.space as Record)?.key, labels, sourceUrl: links?.webui ? `https://${domain}/wiki${links.webui}` : undefined, }) } export const confluenceConnector: ConnectorConfig = { ...confluenceConnectorMeta, listDocuments: async ( accessToken: string, sourceConfig: Record, cursor?: string, syncContext?: Record ): Promise => { const domain = normalizeConfluenceDomainHost(sourceConfig.domain as string) const spaceKeys = parseMultiValue(sourceConfig.spaceKey) const contentType = (sourceConfig.contentType as string) || 'page' const labelFilter = (sourceConfig.labelFilter as string) || '' const maxPages = sourceConfig.maxPages ? Number(sourceConfig.maxPages) : 0 if (spaceKeys.length === 0) { throw new Error('At least one space key is required') } let cloudId = syncContext?.cloudId as string | undefined if (!cloudId) { cloudId = await getConfluenceCloudId(domain, accessToken) if (syncContext) syncContext.cloudId = cloudId } /** * Route through CQL when a label filter is set or when multiple spaces are * selected — the v2 `/spaces/{spaceId}/pages` endpoint is single-space only, * but CQL natively supports `space in (...)`. */ if (labelFilter.trim() || spaceKeys.length > 1) { return listDocumentsViaCql( cloudId, accessToken, domain, spaceKeys, contentType, labelFilter, maxPages, cursor, syncContext ) } const spaceKey = spaceKeys[0] let spaceId = syncContext?.spaceId as string | undefined if (!spaceId) { spaceId = await resolveSpaceId(cloudId, accessToken, spaceKey) if (syncContext) syncContext.spaceId = spaceId } if (contentType === 'all') { return listAllContentTypes( cloudId, accessToken, domain, spaceId, spaceKey, maxPages, cursor, syncContext ) } return listDocumentsV2( cloudId, accessToken, domain, spaceId, spaceKey, contentType, maxPages, cursor, syncContext ) }, getDocument: async ( accessToken: string, sourceConfig: Record, externalId: string, syncContext?: Record ): Promise => { const domain = normalizeConfluenceDomainHost(sourceConfig.domain as string) let cloudId = syncContext?.cloudId as string | undefined if (!cloudId) { cloudId = await getConfluenceCloudId(domain, accessToken) if (syncContext) syncContext.cloudId = cloudId } // Try pages first, fall back to blogposts if not found let page: Record | null = null for (const endpoint of ['pages', 'blogposts']) { const url = `https://api.atlassian.com/ex/confluence/${cloudId}/wiki/api/v2/${endpoint}/${externalId}?body-format=storage` const response = await fetchWithRetry(url, { method: 'GET', headers: { Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, }) if (response.ok) { page = await response.json() break } if (response.status !== 404) { throw new Error(`Failed to get Confluence content: ${response.status}`) } } if (!page) return null const body = page.body as Record | undefined const storage = body?.storage as Record | undefined const rawContent = (storage?.value as string) || '' const plainText = htmlToPlainText(rawContent) const labelMap = await fetchLabelsForPages(cloudId, accessToken, [String(page.id)]) const labels = labelMap.get(String(page.id)) ?? [] const links = page._links as Record | undefined const stub = pageToStub(page, { spaceId: page.spaceId, labels, sourceUrl: links?.webui ? `https://${domain}/wiki${links.webui}` : undefined, }) return { ...stub, content: plainText, contentDeferred: false, } }, validateConfig: async ( accessToken: string, sourceConfig: Record ): Promise<{ valid: boolean; error?: string }> => { const domain = sourceConfig.domain as string const spaceKeys = parseMultiValue(sourceConfig.spaceKey) if (!domain || spaceKeys.length === 0) { return { valid: false, error: 'Domain and at least one space key are required' } } const maxPages = sourceConfig.maxPages as string | undefined if (maxPages && (Number.isNaN(Number(maxPages)) || Number(maxPages) <= 0)) { return { valid: false, error: 'Max pages must be a positive number' } } try { const cloudId = await getConfluenceCloudId(domain, accessToken, VALIDATE_RETRY_OPTIONS) const params = new URLSearchParams() for (const key of spaceKeys) params.append('keys', key) params.append('limit', String(Math.max(spaceKeys.length, 1))) const spaceUrl = `https://api.atlassian.com/ex/confluence/${cloudId}/wiki/api/v2/spaces?${params.toString()}` const response = await fetchWithRetry( spaceUrl, { method: 'GET', headers: { Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, }, VALIDATE_RETRY_OPTIONS ) if (!response.ok) { return { valid: false, error: `Failed to validate spaces: ${response.status}` } } const data = await response.json() const results = (data.results as Array> | undefined) ?? [] const foundKeys = new Set(results.map((r) => String(r.key))) const missing = spaceKeys.filter((k) => !foundKeys.has(k)) if (missing.length > 0) { return { valid: false, error: `Space${missing.length > 1 ? 's' : ''} not found: ${missing.join(', ')}`, } } return { valid: true } } catch (error) { return { valid: false, error: toError(error).message || 'Failed to validate configuration' } } }, mapTags: (metadata: Record): Record => { const result: Record = {} const joined = joinTagArray(metadata.labels) if (joined) result.labels = joined if (metadata.version != null) { const num = Number(metadata.version) if (!Number.isNaN(num)) result.version = num } const lastModified = parseTagDate(metadata.lastModified) if (lastModified) result.lastModified = lastModified return result }, } /** * Lists documents using the v2 API for a single content type (pages or blogposts). */ async function listDocumentsV2( cloudId: string, accessToken: string, domain: string, spaceId: string, spaceKey: string, contentType: string, maxPages: number, cursor?: string, syncContext?: Record ): Promise { const queryParams = new URLSearchParams() queryParams.append('limit', '250') if (cursor) { queryParams.append('cursor', cursor) } const endpoint = contentType === 'blogpost' ? 'blogposts' : 'pages' const url = `https://api.atlassian.com/ex/confluence/${cloudId}/wiki/api/v2/spaces/${spaceId}/${endpoint}?${queryParams.toString()}` logger.info(`Listing ${endpoint} in space ${spaceKey} (ID: ${spaceId})`) 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 list Confluence ${endpoint}`, { status: response.status, error: errorText, }) throw new Error(`Failed to list Confluence ${endpoint}: ${response.status}`) } const data = await response.json() const results = data.results || [] const documents: ExternalDocument[] = results.map((page: Record) => { const links = page._links as Record | undefined return pageToStub(page, { spaceId: page.spaceId, sourceUrl: links?.webui ? `https://${domain}/wiki${links.webui}` : undefined, }) }) let nextCursor: string | undefined const nextLink = (data._links as Record)?.next if (nextLink) { try { nextCursor = new URL(nextLink, 'https://placeholder').searchParams.get('cursor') || undefined } catch { // Ignore malformed URLs } } const totalFetched = ((syncContext?.totalDocsFetched as number) ?? 0) + documents.length if (syncContext) syncContext.totalDocsFetched = totalFetched const hitLimit = maxPages > 0 && totalFetched >= maxPages if (hitLimit && syncContext) syncContext.listingCapped = true return { documents, nextCursor: hitLimit ? undefined : nextCursor, hasMore: hitLimit ? false : Boolean(nextCursor), } } /** * Lists both pages and blogposts using a compound cursor that tracks * pagination state for each content type independently. */ async function listAllContentTypes( cloudId: string, accessToken: string, domain: string, spaceId: string, spaceKey: string, maxPages: number, cursor?: string, syncContext?: Record ): Promise { let pageCursor: string | undefined let blogCursor: string | undefined let pagesDone = false let blogsDone = false if (cursor) { try { const parsed = JSON.parse(cursor) pageCursor = parsed.page blogCursor = parsed.blog pagesDone = parsed.pagesDone === true blogsDone = parsed.blogsDone === true } catch { /** * Older bare-string cursors are no longer emitted; fall through and * restart instead of silently re-listing blogposts from page 0. */ logger.warn('Ignoring unparseable Confluence cursor; restarting listing') } } const results: ExternalDocumentList = { documents: [], hasMore: false } if (!pagesDone) { const pagesResult = await listDocumentsV2( cloudId, accessToken, domain, spaceId, spaceKey, 'page', maxPages, pageCursor, syncContext ) results.documents.push(...pagesResult.documents) pageCursor = pagesResult.nextCursor pagesDone = !pagesResult.hasMore } if (!blogsDone) { const blogResult = await listDocumentsV2( cloudId, accessToken, domain, spaceId, spaceKey, 'blogpost', maxPages, blogCursor, syncContext ) results.documents.push(...blogResult.documents) blogCursor = blogResult.nextCursor blogsDone = !blogResult.hasMore } results.hasMore = !pagesDone || !blogsDone if (results.hasMore) { results.nextCursor = JSON.stringify({ page: pageCursor, blog: blogCursor, pagesDone, blogsDone, }) } return results } /** * Lists documents using CQL search via the v1 API (used when label filtering is enabled). */ async function listDocumentsViaCql( cloudId: string, accessToken: string, domain: string, spaceKeys: string[], contentType: string, labelFilter: string, maxPages: number, cursor?: string, syncContext?: Record ): Promise { const labels = labelFilter .split(',') .map((l) => l.trim()) .filter(Boolean) // Build CQL query let cql = buildSpaceClause(spaceKeys) if (contentType === 'blogpost') { cql += ' AND type="blogpost"' } else if (contentType === 'page' || !contentType) { cql += ' AND type="page"' } // contentType === 'all' — no type filter if (labels.length === 1) { cql += ` AND label="${escapeCql(labels[0])}"` } else if (labels.length > 1) { const labelList = labels.map((l) => `"${escapeCql(l)}"`).join(',') cql += ` AND label in (${labelList})` } const limit = maxPages > 0 ? Math.min(maxPages, 50) : 50 const start = cursor ? Number(cursor) : 0 const queryParams = new URLSearchParams() queryParams.append('cql', cql) queryParams.append('limit', String(limit)) queryParams.append('start', String(start)) queryParams.append('expand', 'version,metadata.labels') const url = `https://api.atlassian.com/ex/confluence/${cloudId}/wiki/rest/api/content/search?${queryParams.toString()}` logger.info(`Searching Confluence via CQL: ${cql}`, { start, limit }) 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 search Confluence via CQL', { status: response.status, error: errorText, }) throw new Error(`Failed to search Confluence via CQL: ${response.status}`) } const data = await response.json() const results = data.results || [] const documents: ExternalDocument[] = results.map((item: Record) => cqlResultToStub(item, domain) ) const totalFetched = ((syncContext?.totalDocsFetched as number) ?? 0) + documents.length if (syncContext) syncContext.totalDocsFetched = totalFetched const hitLimit = maxPages > 0 && totalFetched >= maxPages if (hitLimit && syncContext) syncContext.listingCapped = true const totalSize = (data.totalSize as number) ?? 0 const nextStart = start + results.length const hasMore = !hitLimit && nextStart < totalSize return { documents, nextCursor: hasMore ? String(nextStart) : undefined, hasMore, } } /** * Resolves a Confluence space key to its numeric space ID. */ async function resolveSpaceId( cloudId: string, accessToken: string, spaceKey: string ): Promise { const url = `https://api.atlassian.com/ex/confluence/${cloudId}/wiki/api/v2/spaces?keys=${encodeURIComponent(spaceKey)}&limit=1` const response = await fetchWithRetry(url, { method: 'GET', headers: { Accept: 'application/json', Authorization: `Bearer ${accessToken}`, }, }) if (!response.ok) { throw new Error(`Failed to resolve space key "${spaceKey}": ${response.status}`) } const data = await response.json() const results = data.results || [] if (results.length === 0) { throw new Error(`Space "${spaceKey}" not found`) } return String(results[0].id) }