From 07533859d7e330c16f07f0f8c4c3807e631e5d44 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 10 Sep 2026 15:04:08 -0700 Subject: [PATCH 1/2] fix(knowledge): stop connector sync failures on download-restricted Drive files and macro-only Confluence pages --- .../connectors/confluence/confluence.test.ts | 106 ++++++++++++++++- apps/sim/connectors/confluence/confluence.ts | 110 +++++++++++++++--- .../google-drive/google-drive-errors.ts | 13 ++- .../google-drive/google-drive.test.ts | 75 +++++++++++- .../connectors/google-drive/google-drive.ts | 33 +++++- apps/sim/connectors/slack/slack.test.ts | 25 ++++ apps/sim/connectors/slack/slack.ts | 80 +++++++++++-- .../organization-search-overview.test.ts | 35 ++++++ .../organization-search-overview.ts | 8 +- .../connectors/sync-content-pass.test.ts | 4 +- .../knowledge/connectors/sync-primitives.ts | 2 + 11 files changed, 458 insertions(+), 33 deletions(-) diff --git a/apps/sim/connectors/confluence/confluence.test.ts b/apps/sim/connectors/confluence/confluence.test.ts index d1e5b0b7945..f4c3601b9a9 100644 --- a/apps/sim/connectors/confluence/confluence.test.ts +++ b/apps/sim/connectors/confluence/confluence.test.ts @@ -10,7 +10,9 @@ import { buildLastModifiedClause, confluenceConnector, confluenceStorageToPlainText, + DYNAMIC_CONTENT_SKIP_REASON, escapeCql, + extractConfluenceStorageText, isCurrentContent, preserveConfluenceCallouts, readIncludedLabels, @@ -770,6 +772,77 @@ describe('confluenceStorageToPlainText', () => { expect(confluenceStorageToPlainText(storage)).toBe('Public body') }) + it('keeps text authored inside legacy section and column layouts', () => { + const storage = + '' + + '50%' + + '

Linux Patching

Run the playbook.

' + + '' + + '

Second column

' + + '' + + expect(confluenceStorageToPlainText(storage)).toBe( + 'Linux Patching Run the playbook. Second column' + ) + }) + + it('keeps page properties tables, table-macro bodies, and status labels', () => { + const storage = + '' + + '
OwnerPlatform team
' + + '' + + 'Name' + + '
Filtered row
' + + '' + + '

State: Green' + + 'Approved

' + + expect(confluenceStorageToPlainText(storage)).toBe( + 'Owner Platform team Filtered row State: Approved' + ) + }) + + it('keeps new-editor panel and decision text while dropping app extensions', () => { + const storage = + '' + + 'custom' + + '

Rotate the key quarterly.

' + + '

Rotate the key quarterly.

' + + '' + + 'abc' + + 'DECIDED' + + 'Use Vault' + + '' + + 'remote' + + '

Rendered by an app

' + + expect(confluenceStorageToPlainText(storage)).toBe( + '[CALLOUT] Rotate the key quarterly. Use Vault' + ) + }) + + it('drops template placeholders and task bookkeeping but keeps task text intact', () => { + const storage = + '

Type your summary here

' + + '1u' + + 'incompleteShip it' + + '

Unbelievable

' + + expect(confluenceStorageToPlainText(storage)).toBe('Ship it Unbelievable') + }) + + it('reports whether dynamic content was removed', () => { + expect(extractConfluenceStorageText('

Local

')).toEqual({ + text: 'Local', + droppedDynamicContent: false, + }) + expect( + extractConfluenceStorageText( + '1' + ) + ).toEqual({ text: '', droppedDynamicContent: true }) + }) + it.each(['expand', 'excerpt', 'noformat'])( 'retains the authored content of the %s macro', (name) => { @@ -902,7 +975,36 @@ describe('Confluence permission-scoped content', () => { cloudId: 'cloud-1', mirrorsSourceAcls: true, }) - ).resolves.toMatchObject({ content: '', skippedExistingDisposition: 'replace' }) + ).resolves.toMatchObject({ + content: '', + skippedReason: DYNAMIC_CONTENT_SKIP_REASON, + skippedExistingDisposition: 'replace', + }) + }) + + it('names dynamic-only hub pages distinctly from genuinely empty ones', async () => { + vi.mocked(fetch).mockResolvedValueOnce( + new Response( + JSON.stringify({ + id: 'hub', + version: { number: 2 }, + body: { + storage: { + value: + '' + + 'project = X', + }, + }, + }) + ) + ) + const document = await confluenceConnector.getDocument('token', config, 'hub', { + cloudId: 'cloud-1', + perMemberListing: true, + memberId: 'member-1', + }) + expect(document?.skippedReason).toBe(DYNAMIC_CONTENT_SKIP_REASON) + expect(document?.skippedRetryPolicy).toBe('source-change') }) it('keeps skipped pages retryable when no usable source version is available', async () => { @@ -1015,7 +1117,7 @@ describe('Confluence permission-scoped content', () => { ) const expectedHash = 'mirrorsSourceAcls' in mode || 'perMemberListing' in mode - ? 'confluence:storage-local-body-v1:shared-page:1' + ? 'confluence:storage-local-body-v2:shared-page:1' : 'confluence:view-callouts:shared-page:1' expect(v2.documents[0].contentHash).toBe(expectedHash) diff --git a/apps/sim/connectors/confluence/confluence.ts b/apps/sim/connectors/confluence/confluence.ts index 1b4bb94e446..87986332c5e 100644 --- a/apps/sim/connectors/confluence/confluence.ts +++ b/apps/sim/connectors/confluence/confluence.ts @@ -93,6 +93,7 @@ const INLINE_FORMATTING_TAGS = new Set([ 'var', 'samp', 'time', + 'ac:inline-comment-marker', ]) /** @@ -204,33 +205,98 @@ export function preserveConfluenceCallouts(html: string): string { } const STORAGE_MACRO_SELECTOR = 'ac\\:structured-macro, ac\\:macro' +const ADF_NODE_SELECTOR = 'ac\\:adf-node' +/** Callout macros whose body is prefixed with a semantic label, as on the view path. */ +const LOCAL_CALLOUT_MACROS = new Set(['info', 'note', 'warning', 'tip', 'panel']) +/** + * Macros whose text is authored on the page itself: callouts, expand/excerpt/code + * bodies, legacy `section`/`column` layouts (which wrap the entire body of pages + * built in the old editor), Page Properties (`details`), table-wrapping macros, + * and `status` lozenges. Everything else either resolves another resource + * (include, jira, children, page tree, label reports) or is an app macro, and + * may render differently for each reader. + */ const LOCAL_STORAGE_MACROS = new Set([ - 'info', - 'note', - 'warning', - 'tip', - 'panel', + ...LOCAL_CALLOUT_MACROS, 'expand', 'excerpt', 'code', 'noformat', + 'section', + 'column', + 'details', + 'toc-zone', + 'chart', + 'table-filter', + 'table-chart', + 'table-pivot', + 'table-transformer', + 'table-excerpt', + 'table-plus', + 'status', ]) +/** New-editor nodes stored as ADF whose content is authored on the page. */ +const LOCAL_ADF_NODE_TYPES = new Set(['panel', 'decision-list', 'decision-item']) +/** ADF nodes rendered by a Forge or Connect app; their output is resolved elsewhere. */ +const APP_ADF_NODE_TYPES = new Set(['extension', 'bodiedExtension', 'inlineExtension']) +/** Storage-format bookkeeping that is never page prose. */ +const STORAGE_NOISE_SELECTOR = [ + 'ac\\:parameter', + 'ac\\:default-parameter', + 'ac\\:adf-attribute', + 'ac\\:adf-fallback', + 'ac\\:placeholder', + 'ac\\:task-id', + 'ac\\:task-uuid', + 'ac\\:task-status', + 'script', + 'style', +].join(', ') + +/** Recorded when a scoped page holds nothing but content resolved from elsewhere. */ +export const DYNAMIC_CONTENT_SKIP_REASON = + 'Page only contains dynamic content (child lists, includes, or app macros) that Search cannot index' + +export interface ConfluenceStorageText { + text: string + /** True when at least one non-local macro or app node was removed. */ + droppedDynamicContent: boolean +} /** * Search authorizes the containing page, not content expanded from another * resource. Read authored storage text and known local macro bodies only; * inclusion and third-party macros may render differently for each reader. */ -export function confluenceStorageToPlainText(storage: string): string { +export function extractConfluenceStorageText(storage: string): ConfluenceStorageText { const $ = cheerio.load( storage, { xml: { xmlMode: false, recognizeCDATA: true, recognizeSelfClosing: true } }, false ) - $('ac\\:adf-extension').remove() + let droppedDynamicContent = false + + for (const element of $(ADF_NODE_SELECTOR).toArray().reverse()) { + const node = $(element) + const type = node.attr('type') ?? '' + if (!LOCAL_ADF_NODE_TYPES.has(type)) { + if (APP_ADF_NODE_TYPES.has(type)) droppedDynamicContent = true + node.remove() + continue + } + const panelType = node.children('ac\\:adf-attribute[key="panel-type"]').text().trim() + node.children('ac\\:adf-attribute, ac\\:adf-fallback').remove() + const body = extractBlockJoinedText($, node) + const label = + type === 'panel' + ? (CALLOUT_LABELS[panelType === 'info' ? 'information' : panelType] ?? '[CALLOUT]') + : '' + node.replaceWith($('

').text([label, body].filter(Boolean).join(' '))) + } $(STORAGE_MACRO_SELECTOR).each((_, element) => { if (!LOCAL_STORAGE_MACROS.has($(element).attr('ac:name') ?? '')) { + droppedDynamicContent = true $(element).remove() } }) @@ -248,13 +314,23 @@ export function confluenceStorageToPlainText(storage: string): string { ? title ? `[CALLOUT: ${title}]` : '[CALLOUT]' - : CALLOUT_LABELS[name === 'info' ? 'information' : name] + : LOCAL_CALLOUT_MACROS.has(name) + ? CALLOUT_LABELS[name === 'info' ? 'information' : name] + : '' const text = [label, name === 'panel' ? '' : title, body].filter(Boolean).join(' ') macro.replaceWith($('

').text(text)) } - $('ac\\:parameter, ac\\:default-parameter, script, style').remove() - return extractBlockJoinedText($, $.root()).replace(/\s+/g, ' ').trim() + $(STORAGE_NOISE_SELECTOR).remove() + return { + text: extractBlockJoinedText($, $.root()).replace(/\s+/g, ' ').trim(), + droppedDynamicContent, + } +} + +/** Plain text of a storage-format body; see {@link extractConfluenceStorageText}. */ +export function confluenceStorageToPlainText(storage: string): string { + return extractConfluenceStorageText(storage).text } function usesPermissionScopedContent(syncContext?: Record): boolean { @@ -313,7 +389,7 @@ export function readIncludedLabels(page: Record): string[] { * ordinary knowledge bases retain their existing rendered representation. */ const CONTENT_REPRESENTATION = 'view-callouts' -const SCOPED_CONTENT_REPRESENTATION = 'storage-local-body-v1' +const SCOPED_CONTENT_REPRESENTATION = 'storage-local-body-v2' /** * Produces a canonical metadata stub with a deterministic contentHash that @@ -690,9 +766,8 @@ export const confluenceConnector: ConnectorConfig = { throw new Error(`Confluence content is missing its ${bodyFormat} body`) } const rawContent = representation.value - const plainText = scopedContent - ? confluenceStorageToPlainText(rawContent) - : htmlToPlainText(preserveConfluenceCallouts(rawContent)) + const scoped = scopedContent ? extractConfluenceStorageText(rawContent) : null + const plainText = scoped ? scoped.text : htmlToPlainText(preserveConfluenceCallouts(rawContent)) const links = page._links as Record | undefined const stub = pageToStub( @@ -707,7 +782,12 @@ export const confluenceConnector: ConnectorConfig = { if (!plainText.trim()) { return { - ...markSkipped(stub, 'Document contains no extractable text'), + ...markSkipped( + stub, + scoped?.droppedDynamicContent + ? DYNAMIC_CONTENT_SKIP_REASON + : 'Document contains no extractable text' + ), skippedExistingDisposition: 'replace', } } diff --git a/apps/sim/connectors/google-drive/google-drive-errors.ts b/apps/sim/connectors/google-drive/google-drive-errors.ts index b57cd059b27..db9c993b5e5 100644 --- a/apps/sim/connectors/google-drive/google-drive-errors.ts +++ b/apps/sim/connectors/google-drive/google-drive-errors.ts @@ -20,7 +20,18 @@ const PERMISSION_REASONS = new Set([ 'insufficientFilePermissions', 'teamDriveMembershipRequired', ]) -const POLICY_REASONS = new Set(['domainPolicy', 'download_restricted_for_revision']) +/** + * Owner- or admin-imposed restrictions on an otherwise readable file. The + * credential is valid, so these are not authorization failures; `cannotExportFile` + * and `cannotDownloadFile` are what Drive returns when the owner disabled + * download, print, and copy for viewers. + */ +const POLICY_REASONS = new Set([ + 'domainPolicy', + 'download_restricted_for_revision', + 'cannotDownloadFile', + 'cannotExportFile', +]) const UNSUPPORTED_EXPORT_REASONS = new Set(['fileNotDownloadable', 'fileNotExportable']) const QUOTA_REASONS = new Set(['dailyLimitExceeded', 'quotaExceeded']) const RATE_LIMIT_REASONS = new Set([ diff --git a/apps/sim/connectors/google-drive/google-drive.test.ts b/apps/sim/connectors/google-drive/google-drive.test.ts index f538e5b81bd..06444b02cf4 100644 --- a/apps/sim/connectors/google-drive/google-drive.test.ts +++ b/apps/sim/connectors/google-drive/google-drive.test.ts @@ -41,7 +41,10 @@ afterEach(() => { vi.unstubAllGlobals() }) -import { googleDriveConnector } from '@/connectors/google-drive/google-drive' +import { + DOWNLOAD_RESTRICTED_SKIP_REASON, + googleDriveConnector, +} from '@/connectors/google-drive/google-drive' import { GoogleDriveApiError, readGoogleDriveApiError, @@ -458,6 +461,8 @@ describe('Google Drive API error parsing', () => { ['insufficientFilePermissions', 'permission'], ['appNotAuthorizedToFile', 'permission'], ['domainPolicy', 'policy'], + ['cannotDownloadFile', 'policy'], + ['cannotExportFile', 'policy'], ['fileNotExportable', 'unsupported_export'], ['dailyLimitExceeded', 'quota'], ['rateLimitExceeded', 'transient'], @@ -564,6 +569,72 @@ describe('Google Drive API error parsing', () => { }) }) +describe('Google Drive download-restricted files', () => { + beforeEach(() => { + vi.clearAllMocks() + vi.stubGlobal('fetch', mockFetch) + }) + + const restricted = () => fileMetadata({ capabilities: { canDownload: false } }) + + it('always asks Drive whether the file can be downloaded', async () => { + mockFetch.mockResolvedValueOnce(jsonResponse({ files: [] })) + await googleDriveConnector.listDocuments('token', {}, undefined, {}) + + const fields = decodeURIComponent(String(mockFetch.mock.calls[0][0])) + expect(fields).toContain('capabilities(canDownload)') + expect(fields).not.toContain('permissions(') + }) + + it.each([ + ['a workspace crawl', {}], + ['a per-member listing', { perMemberListing: true }], + ])('skips a restricted file at listing time in %s', async (_label, syncContext) => { + mockFetch.mockResolvedValueOnce(jsonResponse({ files: [restricted()] })) + const page = await googleDriveConnector.listDocuments('token', {}, undefined, syncContext) + + expect(page.documents).toHaveLength(1) + expect(page.documents[0].skippedReason).toBe(DOWNLOAD_RESTRICTED_SKIP_REASON) + expect(mockFetch).toHaveBeenCalledTimes(1) + }) + + it('lists a downloadable file as an ordinary deferred stub', async () => { + mockFetch.mockResolvedValueOnce( + jsonResponse({ files: [fileMetadata({ capabilities: { canDownload: true } })] }) + ) + const page = await googleDriveConnector.listDocuments('token', {}, undefined, {}) + + expect(page.documents[0].skippedReason).toBeUndefined() + expect(page.documents[0].contentDeferred).toBe(true) + }) + + it('skips hydration without calling export when metadata says the file is restricted', async () => { + mockFetch.mockResolvedValueOnce(jsonResponse(restricted())) + const document = await googleDriveConnector.getDocument('token', {}, FILE_ID) + + expect(document?.skippedReason).toBe(DOWNLOAD_RESTRICTED_SKIP_REASON) + expect(document?.contentDeferred).toBe(false) + expect(mockFetch).toHaveBeenCalledTimes(1) + }) + + it('reports a restricted file from the change feed as a skipped upsert', async () => { + mockFetch.mockResolvedValueOnce( + jsonResponse({ + changes: [{ changeType: 'file', fileId: FILE_ID, file: restricted() }], + newStartPageToken: '2', + }) + ) + const page = await googleDriveConnector.listChanges?.('token', {}, '1', {}) + + expect(page?.changes).toHaveLength(1) + const change = page?.changes[0] + expect(change?.kind).toBe('upsert') + if (change?.kind === 'upsert') { + expect(change.document.skippedReason).toBe(DOWNLOAD_RESTRICTED_SKIP_REASON) + } + }) +}) + describe('Google Drive metadata hydration', () => { beforeEach(() => { vi.clearAllMocks() @@ -606,6 +677,8 @@ describe('Google Drive export failures', () => { 403, ], ['domainPolicy', 'The domain administrators have disabled Drive apps.', 403], + ['cannotExportFile', 'This file cannot be exported by the user.', 403], + ['cannotDownloadFile', 'This file cannot be downloaded by the user.', 403], ['fileNotExportable', 'This file cannot be exported.', 403], ])( 'propagates recoverable %s failures instead of persisting a sticky same-hash skip', diff --git a/apps/sim/connectors/google-drive/google-drive.ts b/apps/sim/connectors/google-drive/google-drive.ts index fdca1d94b87..88d9413f5a4 100644 --- a/apps/sim/connectors/google-drive/google-drive.ts +++ b/apps/sim/connectors/google-drive/google-drive.ts @@ -73,7 +73,21 @@ const SHORTCUT_FETCH_CONCURRENCY = 8 const DRIVE_METADATA_MAX_BYTES = 1024 * 1024 const DRIVE_PAGE_MAX_BYTES = 16 * 1024 * 1024 const DRIVE_FILE_FIELDS = - 'id,name,mimeType,modifiedTime,createdTime,webViewLink,owners,size,starred,trashed,parents,shortcutDetails(targetId,targetMimeType,targetResourceKey)' + 'id,name,mimeType,modifiedTime,createdTime,webViewLink,owners,size,starred,trashed,parents,shortcutDetails(targetId,targetMimeType,targetResourceKey),capabilities(canDownload)' + +/** + * Recorded on a listed file whose owner disabled download, print, and copy for + * viewers. Drive rejects both `files.export` and `files.get?alt=media` for such a + * file (`cannotExportFile` / `cannotDownloadFile`), so the crawl skips the fetch + * and surfaces the restriction instead of failing hydration on every sync. + */ +export const DOWNLOAD_RESTRICTED_SKIP_REASON = + 'The file owner has disabled downloading for viewers, so its content cannot be indexed' + +/** True when the file's metadata says the acting credential cannot read its bytes. */ +function isDownloadRestricted(file: DriveFile): boolean { + return file.capabilities?.canDownload === false +} const FOLDER_MIME_TYPE = 'application/vnd.google-apps.folder' @@ -427,6 +441,13 @@ function driveChangeToExternal( if (change.removed || !file || !isFileInScope(file, sourceConfig)) { return { kind: 'removed', externalId } } + if (isDownloadRestricted(file)) { + return { + kind: 'upsert', + externalId, + document: markSkipped(fileToStub(file), DOWNLOAD_RESTRICTED_SKIP_REASON), + } + } return { kind: 'upsert', externalId, @@ -731,7 +752,7 @@ async function readDriveFile( resourceKey?: string, permissions = false ): Promise { - const fields = `${DRIVE_FILE_FIELDS}${permissions ? `,permissions(${DRIVE_PERMISSION_FIELDS}),capabilities(canDownload)` : ''}` + const fields = `${DRIVE_FILE_FIELDS}${permissions ? `,permissions(${DRIVE_PERMISSION_FIELDS})` : ''}` const response = await fetchGoogleDriveWithRetry( `https://www.googleapis.com/drive/v3/files/${encodeURIComponent(fileId)}?fields=${encodeURIComponent(fields)}&supportsAllDrives=true`, { method: 'GET', headers: driveRequestHeaders(accessToken, fileId, resourceKey) } @@ -860,6 +881,8 @@ async function listedFileToDocument( } const contentFile = target ?? file if (!matchesFileType((sourceConfig.fileType as string) || 'all', contentFile)) return null + if (isDownloadRestricted(contentFile)) + return markSkipped(fileToStub(file, acl, target), DOWNLOAD_RESTRICTED_SKIP_REASON) return stubOrSkipBySize( fileToStub(file, acl, target), Number(contentFile.size) || undefined, @@ -1069,7 +1092,7 @@ const listGoogleDriveDocuments: ConnectorConfig['listDocuments'] = async ( * crawl would pull a permission array per file and discard it. */ fields: `kind,nextPageToken,incompleteSearch,files(${DRIVE_FILE_FIELDS}${ - aclContext ? `,permissions(${DRIVE_PERMISSION_FIELDS}),capabilities(canDownload)` : '' + aclContext ? `,permissions(${DRIVE_PERMISSION_FIELDS})` : '' })`, supportsAllDrives: 'true', includeItemsFromAllDrives: 'true', @@ -1252,6 +1275,10 @@ export const googleDriveConnector: ConnectorConfig = { } } + if (isDownloadRestricted(contentFile)) { + return markSkipped(stub, DOWNLOAD_RESTRICTED_SKIP_REASON) + } + try { const payload = await fetchFilePayload(accessToken, contentFile, resourceKey) if (!payload.content.trim() && !payload.sourceFile?.bytes.length) { diff --git a/apps/sim/connectors/slack/slack.test.ts b/apps/sim/connectors/slack/slack.test.ts index 1bf8f9b18b4..3353e691137 100644 --- a/apps/sim/connectors/slack/slack.test.ts +++ b/apps/sim/connectors/slack/slack.test.ts @@ -550,6 +550,31 @@ describe('Slack incomplete and unsafe provider responses', () => { } ) + it.each([ + { code: 'ratelimited', status: 429, category: 'rate_limit' }, + { code: 'token_revoked', status: 401, category: 'authorization' }, + { code: 'missing_scope', status: 403, category: 'authorization' }, + { code: 'internal_error', status: 503, category: 'provider_unavailable' }, + { code: 'invalid_arguments', status: 400, category: 'request_rejected' }, + ])('classifies the $code envelope as HTTP $status $category', async ({ code, ...expected }) => { + failure = (call) => + call.method === 'conversations.replies' ? { ok: false, error: code } : undefined + await expect(slackConnector.getDocument('alice', {}, id(GENERAL.id))).rejects.toMatchObject({ + name: 'SlackApiError', + code, + ...expected, + }) + }) + + it('keeps the HTTP status of a non-OK response for failure classification', async () => { + fetchMock.mockImplementationOnce(async () => new Response('forbidden', { status: 403 })) + await expect(slackConnector.getDocument('alice', {}, id(GENERAL.id))).rejects.toMatchObject({ + name: 'ConnectorSourceError', + status: 403, + message: expect.stringMatching(/^Slack [a-z.]+ failed with HTTP 403$/), + }) + }) + it('propagates Slack envelope throttling so the sync scheduler can cool down', async () => { failure = (call) => call.method === 'conversations.replies' ? { ok: false, error: 'ratelimited' } : undefined diff --git a/apps/sim/connectors/slack/slack.ts b/apps/sim/connectors/slack/slack.ts index 0f99b9ed2a1..22c2ace4b29 100644 --- a/apps/sim/connectors/slack/slack.ts +++ b/apps/sim/connectors/slack/slack.ts @@ -15,6 +15,10 @@ import { readSlackConversationSetting as readConversationSetting, } from '@/connectors/slack/config' import { DEFAULT_MAX_MESSAGES, slackConnectorMeta } from '@/connectors/slack/meta' +import { + ConnectorSourceError, + type ConnectorSourceFailureCategory, +} from '@/connectors/source-error' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { BoundedLines, @@ -95,16 +99,70 @@ interface SlackListingCursor { scanned: number } +interface SlackCodeClassification { + /** The HTTP status Slack would have used had it not answered 200 with an error envelope. */ + status: number + category: ConnectorSourceFailureCategory +} + +/** + * Slack answers HTTP 200 with `ok: false` and a machine-readable code. Mapping + * the known codes onto the shared failure categories lets the sync engine and + * the stored document error tell a revoked token from a missing thread from a + * throttle, instead of every envelope error reading as an unknown failure. + */ +const SLACK_CODE_CLASSIFICATIONS: ReadonlyArray<[ReadonlySet, SlackCodeClassification]> = [ + [new Set(['ratelimited']), { status: 429, category: 'rate_limit' }], + [ + new Set(['invalid_auth', 'token_revoked', 'token_expired', 'account_inactive', 'not_authed']), + { status: 401, category: 'authorization' }, + ], + [ + new Set(['missing_scope', 'access_denied', 'restricted_action', 'ekm_access_denied']), + { status: 403, category: 'authorization' }, + ], + [ + new Set([ + 'channel_not_found', + 'not_in_channel', + 'channel_is_limited_access', + 'thread_not_found', + 'message_not_found', + ]), + { status: 404, category: 'source_unavailable' }, + ], + [ + new Set(['service_unavailable', 'internal_error', 'fatal_error', 'request_timeout']), + { status: 503, category: 'provider_unavailable' }, + ], +] + +/** Unknown codes are treated as a rejected request; the status alone drives the diagnostic. */ +const UNCLASSIFIED_SLACK_CODE: SlackCodeClassification = { + status: 400, + category: 'request_rejected', +} + +function classifySlackCode(code: string): SlackCodeClassification { + for (const [codes, classification] of SLACK_CODE_CLASSIFICATIONS) { + if (codes.has(code)) return classification + } + return UNCLASSIFIED_SLACK_CODE +} + /** Slack's HTTP-200 errors still retain their machine-readable provider code. */ -class SlackApiError extends Error { +class SlackApiError extends ConnectorSourceError { + readonly code: string + readonly method: string + readonly headers?: Headers readonly rateLimited: boolean - constructor( - readonly code: string, - readonly method: string, - readonly headers?: Headers - ) { - super(`Slack ${method} failed: ${code}`) + constructor(code: string, method: string, headers?: Headers) { + const { status, category } = classifySlackCode(code) + super(`Slack ${method} failed: ${code}`, status, category) this.name = 'SlackApiError' + this.code = code + this.method = method + this.headers = headers this.rateLimited = code === 'ratelimited' } } @@ -120,7 +178,13 @@ async function slackApiGet( { headers: { Authorization: `Bearer ${accessToken}`, Accept: 'application/json' } }, retryOptions ) - if (!response.ok) throw new Error(`Slack ${method} failed with HTTP ${response.status}`) + /** Retries are exhausted by now; the status must survive so the failure can be classified. */ + if (!response.ok) { + throw new ConnectorSourceError( + `Slack ${method} failed with HTTP ${response.status}`, + response.status + ) + } const data = await readResponseJsonWithLimit(response, { maxBytes: MAX_RESPONSE_BYTES, label: `Slack ${method} response`, diff --git a/apps/sim/lib/knowledge/application/organization-search-overview.test.ts b/apps/sim/lib/knowledge/application/organization-search-overview.test.ts index 043fe0f5f8d..6bb20aa3796 100644 --- a/apps/sim/lib/knowledge/application/organization-search-overview.test.ts +++ b/apps/sim/lib/knowledge/application/organization-search-overview.test.ts @@ -29,6 +29,31 @@ vi.mock('@/lib/sim-search/connectors', () => ({ import { organizationSearchOverviewSchema } from '@/lib/api/contracts/knowledge/connectors' import { knowledgeOperations } from '@/lib/knowledge/application/operations' import { readOrganizationSearchOverview } from '@/lib/knowledge/application/organization-search-overview' +import { SOURCE_CONTENT_ERROR } from '@/lib/knowledge/connectors/sync-limits' + +/** The global drizzle mock nests fragments as params; flatten one for inspection. */ +function renderFragment(fragment: unknown): { sql: string; params: unknown[] } { + if (!fragment || typeof fragment !== 'object') return { sql: '', params: [] } + if ('conditions' in fragment && Array.isArray(fragment.conditions)) { + const parts = fragment.conditions.map(renderFragment) + return { + sql: parts.map((part) => part.sql).join(' '), + params: parts.flatMap((part) => part.params), + } + } + const rendered = (fragment as { toSQL?: () => { sql: string; params: unknown[] } }).toSQL?.() + if (!rendered) return { sql: '', params: [] } + const params: unknown[] = [] + let sqlText = rendered.sql + for (const param of rendered.params) { + if (param && typeof param === 'object' && 'toSQL' in param) { + const nested = renderFragment(param) + sqlText += ` ${nested.sql}` + params.push(...nested.params) + } else params.push(param) + } + return { sql: sqlText, params } +} const principal = { kind: 'session', userId: 'admin', sessionId: 'session' } as const const input = { organizationId: 'organization' } @@ -82,6 +107,16 @@ describe('organization Search administration overview', () => { expect(result.providers[0]).toMatchObject({ status: 'needs_attention', issue }) expect(JSON.stringify(result)).not.toContain('private provider response') }) + it('does not count the per-document relisting marker as a member account error', async () => { + queueTableRows(member, [{ role: 'admin' }]) + queueTableRows(knowledgeConnector, [{ ...health }]) + await readOrganizationSearchOverview.execute({ principal, input }) + const rendered = dbChainMockFns.where.mock.calls.flatMap((call) => call.map(renderFragment)) + const memberErrorClause = rendered.find((fragment) => fragment.sql.includes("'suspended'")) + expect(memberErrorClause).toBeDefined() + expect(memberErrorClause?.sql).toContain('IS NOT NULL AND ? <> ?') + expect(memberErrorClause?.params).toContain(SOURCE_CONTENT_ERROR) + }) it('keeps recovery observable while a previous error remains visible', async () => { queueTableRows(member, [{ role: 'admin' }]) queueTableRows(knowledgeConnector, [{ ...health, hasError: true, hasIndexing: true }]) diff --git a/apps/sim/lib/knowledge/application/organization-search-overview.ts b/apps/sim/lib/knowledge/application/organization-search-overview.ts index 70025fc809b..eb58334a776 100644 --- a/apps/sim/lib/knowledge/application/organization-search-overview.ts +++ b/apps/sim/lib/knowledge/application/organization-search-overview.ts @@ -15,6 +15,7 @@ import { SOURCE_ACL_MAX_AGE_MS } from '@/lib/knowledge/access/freshness' import { defineAuthorizedKnowledgeUseCase } from '@/lib/knowledge/application/authorized-knowledge-use-case' import { resolveKnowledgeOwnerContext } from '@/lib/knowledge/application/contexts' import { knowledgeOperations } from '@/lib/knowledge/application/operations' +import { SOURCE_CONTENT_ERROR } from '@/lib/knowledge/connectors/sync-limits' import { MAX_SEARCH_SOURCE_PROVIDER_TYPES } from '@/lib/knowledge/constants' import { canConnectWithDefaults, SEARCH_SOURCE_TYPES } from '@/lib/sim-search/connectors' @@ -100,6 +101,11 @@ export const readOrganizationSearchOverview = defineAuthorizedKnowledgeUseCase({ )) )` const cutoff = sql`statement_timestamp() - (${SOURCE_ACL_MAX_AGE_MS} * interval '1 millisecond')` + /** + * A member whose last run only had per-document content failures carries + * {@link SOURCE_CONTENT_ERROR} as a marker so its next run lists fully; the + * member itself is healthy and the documents are reported separately. + */ const hasMemberError = exists( db .select({ id: knowledgeConnectorMember.id }) @@ -110,7 +116,7 @@ export const readOrganizationSearchOverview = defineAuthorizedKnowledgeUseCase({ sql`( ${knowledgeConnectorMember.status} = 'suspended' OR (${knowledgeConnectorMember.status} = 'active' AND ( - ${knowledgeConnectorMember.lastError} IS NOT NULL + (${knowledgeConnectorMember.lastError} IS NOT NULL AND ${knowledgeConnectorMember.lastError} <> ${SOURCE_CONTENT_ERROR}) OR ${knowledgeConnectorMember.consecutiveFailures} > 0 OR coalesce(${knowledgeConnectorMember.memberSyncedThrough}, ${knowledgeConnectorMember.lastCompleteListingAt}, ${knowledgeConnectorMember.createdAt}) < ${cutoff} )) diff --git a/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts b/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts index cec28ccf473..de2d738b106 100644 --- a/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts @@ -655,7 +655,7 @@ describe('Confluence empty content through the shared content pass', () => { it('explicitly rehydrates a skipped source whose version is unchanged', async () => { sourceBody = { value: '

Local content is rechecked

' } const { result, hydrate } = await runPass({ - existing: { ...EXISTING, contentHash: 'confluence:storage-local-body-v1:page:3' }, + existing: { ...EXISTING, contentHash: 'confluence:storage-local-body-v2:page:3' }, access: 'admin', readCurrent: true, forceRehydrate: true, @@ -669,7 +669,7 @@ describe('Confluence empty content through the shared content pass', () => { hydrationVersion = 3 await runPass({ existing: EXISTING, readCurrent: true, access: 'admin' }) const skipped = contentWrite() - expect(skipped.contentHash).toBe('confluence:storage-local-body-v1:page:3') + expect(skipped.contentHash).toBe('confluence:storage-local-body-v2:page:3') hydrationVersion = undefined sourceBody = { value: '

Current version content

' } diff --git a/apps/sim/lib/knowledge/connectors/sync-primitives.ts b/apps/sim/lib/knowledge/connectors/sync-primitives.ts index 5dc93a8cc4a..c2191ce5d35 100644 --- a/apps/sim/lib/knowledge/connectors/sync-primitives.ts +++ b/apps/sim/lib/knowledge/connectors/sync-primitives.ts @@ -1152,6 +1152,7 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise connectorId, externalId: deferredOps[i].extDoc.externalId, error: diagnostic?.message ?? SOURCE_CONTENT_ERROR, + cause: toError(outcome.reason).message, diagnostic, }) } @@ -1275,6 +1276,7 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise connectorId, externalId: batch[j].extDoc.externalId, error: diagnostic?.message ?? SOURCE_CONTENT_ERROR, + cause: toError(outcome.reason).message, diagnostic, }) } From 2e4501f5493f2ab463642a7a7ffd004033bc79be Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 10 Sep 2026 15:16:08 -0700 Subject: [PATCH 2/2] fix(knowledge): drop raw error message from hydration failure logs --- apps/sim/lib/knowledge/connectors/sync-primitives.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/apps/sim/lib/knowledge/connectors/sync-primitives.ts b/apps/sim/lib/knowledge/connectors/sync-primitives.ts index c2191ce5d35..5dc93a8cc4a 100644 --- a/apps/sim/lib/knowledge/connectors/sync-primitives.ts +++ b/apps/sim/lib/knowledge/connectors/sync-primitives.ts @@ -1152,7 +1152,6 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise connectorId, externalId: deferredOps[i].extDoc.externalId, error: diagnostic?.message ?? SOURCE_CONTENT_ERROR, - cause: toError(outcome.reason).message, diagnostic, }) } @@ -1276,7 +1275,6 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise connectorId, externalId: batch[j].extDoc.externalId, error: diagnostic?.message ?? SOURCE_CONTENT_ERROR, - cause: toError(outcome.reason).message, diagnostic, }) }