diff --git a/apps/sim/connectors/slack/slack.test.ts b/apps/sim/connectors/slack/slack.test.ts index 84a4f8ee54d..1bf8f9b18b4 100644 --- a/apps/sim/connectors/slack/slack.test.ts +++ b/apps/sim/connectors/slack/slack.test.ts @@ -410,6 +410,66 @@ describe('Slack thread indexing through provider APIs', () => { }) }) +describe('Slack threads without indexable text', () => { + beforeEach(() => { + pageSize = 1 + channels = [ + { + channel: GENERAL, + readers: ['alice'], + messages: [{ ...root(''), thread_ts: ROOT }], + replies: { [ROOT]: [{ ...root(''), thread_ts: ROOT }, reply('')] }, + }, + ] + }) + + it('explicitly skips a listed thread only after reading all its reply pages', async () => { + const listed = await listAll('alice') + expect(listed.documents).toHaveLength(1) + const document = await slackConnector.getDocument('alice', {}, id(GENERAL.id), listed.context) + expect(document).toMatchObject({ + externalId: listed.documents[0].externalId, + content: '', + contentDeferred: false, + skippedReason: 'Document contains no extractable text', + skippedExistingDisposition: 'replace', + metadata: { messageCount: 0, rootTs: ROOT, channelId: GENERAL.id, teamId: TEAM }, + }) + expect(calls.filter((call) => call.method === 'conversations.replies')).toHaveLength(2) + expect(calls.some((call) => call.method === 'chat.getPermalink')).toBe(false) + }) + + it('indexes a later reply edit even when the root and reply count have not changed', async () => { + const empty = await slackConnector.getDocument('alice', {}, id(GENERAL.id)) + channels[0].replies[ROOT][1] = reply('Orion has a launch date') + const document = await slackConnector.getDocument('alice', {}, id(GENERAL.id)) + expect(document?.externalId).toBe(empty?.externalId) + expect(document?.content).toContain('Orion has a launch date') + expect(document?.contentHash).not.toBe(empty?.contentHash) + expect(document?.skippedReason).toBeUndefined() + }) + + it('does not classify a missing root as verified empty content', async () => { + replacement = (call) => + call.method === 'conversations.replies' ? { ok: true, messages: [] } : undefined + expect(await slackConnector.getDocument('alice', {}, id(GENERAL.id))).toBeNull() + }) + + it.each([ + [{ ok: true, messages: [reply('')], is_limited: true }, 'only part'], + [{ ok: true, messages: [reply('')], has_more: true }, 'continuation cursor'], + [{ ok: false, error: 'missing_scope' }, 'missing_scope'], + [{ ok: true }, 'invalid message page'], + ])( + 'does not skip an empty thread when a later page is incomplete: %j', + async (response, error) => { + replacement = (call) => + call.method === 'conversations.replies' && call.params.has('cursor') ? response : undefined + await expect(slackConnector.getDocument('alice', {}, id(GENERAL.id))).rejects.toThrow(error) + } + ) +}) + describe('Slack incomplete and unsafe provider responses', () => { it('does not complete a member observation when channel access disappears between history pages', async () => { pageSize = 1 diff --git a/apps/sim/connectors/slack/slack.ts b/apps/sim/connectors/slack/slack.ts index 32d3128e181..0f99b9ed2a1 100644 --- a/apps/sim/connectors/slack/slack.ts +++ b/apps/sim/connectors/slack/slack.ts @@ -20,6 +20,7 @@ import { BoundedLines, CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, ConnectorFileTooLargeError, + markSkipped, parseDefaultedUnlimitedSafeInteger, parseMultiValue, parseTagDate, @@ -693,22 +694,14 @@ async function getDocument( cursor = continuation } if (!exhausted) throw new Error(`Slack thread exceeds ${MAX_THREAD_PAGES} reply pages`) - if (!root || lines.count === 0) return null - const link = await slackApiGet('chat.getPermalink', accessToken, { - channel: channelId, - message_ts: rootTs, - }) - if (typeof link.permalink !== 'string' || !link.permalink.startsWith('https://')) { - throw new Error('Slack did not return a message permalink') - } - const content = lines.join() - return { + if (!root) return null + const content = lines.count > 0 ? lines.join() : '' + const document: ExternalDocument = { externalId, title: messageTitle(channel, root), content, contentDeferred: false, mimeType: 'text/plain', - sourceUrl: link.permalink, contentHash: `slack-content:v4:${createHash('sha256').update(content).digest('hex')}`, metadata: { channelName: channel.name, @@ -720,6 +713,21 @@ async function getDocument( lastActivity: new Date(Number(lastActivity) * 1000).toISOString(), }, } + /** Only a fully read thread can authoritatively replace previously indexed text with a skip. */ + if (lines.count === 0) { + return { + ...markSkipped(document, 'Document contains no extractable text'), + skippedExistingDisposition: 'replace', + } + } + const link = await slackApiGet('chat.getPermalink', accessToken, { + channel: channelId, + message_ts: rootTs, + }) + if (typeof link.permalink !== 'string' || !link.permalink.startsWith('https://')) { + throw new Error('Slack did not return a message permalink') + } + return { ...document, sourceUrl: link.permalink } } catch (error) { if ( error instanceof SlackApiError && diff --git a/apps/sim/lib/knowledge/__integration__/slack-empty-threads.integration.ts b/apps/sim/lib/knowledge/__integration__/slack-empty-threads.integration.ts new file mode 100644 index 00000000000..2ec84a824a7 --- /dev/null +++ b/apps/sim/lib/knowledge/__integration__/slack-empty-threads.integration.ts @@ -0,0 +1,233 @@ +/** Real sync jobs, PostgreSQL, storage, indexing and authorized search; Slack and embedding responses are synthetic. */ +import { mkdtempSync } from 'node:fs' +import { rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { db } from '@sim/db' +import { + document, + embedding, + knowledgeConnector, + organization, + user, + workspace, +} from '@sim/db/schema' +import { generateId } from '@sim/utils/id' +import { and, eq, inArray } from 'drizzle-orm' +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest' + +const fixture = vi.hoisted(() => ({ storageRoot: '', embeddingCalls: 0 })) +vi.mock('@/lib/uploads/core/setup.server', () => ({ + get UPLOAD_DIR_SERVER() { + return fixture.storageRoot + }, +})) +vi.mock('@/lib/embeddings', async () => ({ + ...(await import('@/lib/embeddings/client')), + assertKnowledgeEmbeddingCapacity: async () => {}, + embedKnowledge: async (texts: string[]) => { + fixture.embeddingCalls++ + return { + embeddings: texts.map(() => [1, ...Array(1535).fill(0)]), + totalTokens: texts.length, + billableTokens: 0, + isBYOK: true, + modelName: 'text-embedding-3-small', + pricingId: 'text-embedding-3-small', + } + }, +})) + +import { resolveBillingAttribution } from '@/lib/billing/core/billing-attribution' +import { + createKnowledgeAclFixtureIds, + seedKnowledgeAclFixture, +} from '@/lib/knowledge/__integration__/seed-source-access-fixture' +import { searchKnowledge } from '@/lib/knowledge/application/search' +import * as connectorTokens from '@/lib/knowledge/connectors/access-token' +import { executeConnectorSyncJob } from '@/background/knowledge-connector-sync' + +const TEAM = 'T0FIXTURE' +const CHANNEL = 'C0GENERAL' +const ROOT = '1700000100.000100' +const REPLY = '1700000200.000100' +const EXTERNAL_ID = `slack:v4:${TEAM}:${CHANNEL}:${ROOT}` +const EMPTY_REASON = 'Document contains no extractable text' + +describe('Slack empty threads through sync jobs, indexing and search', () => { + const ids = createKnowledgeAclFixtureIds() + const documentId = generateId() + let billing: Awaited> + let replyText = '' + let incomplete = false + let missingRoot = false + const channel = { id: CHANNEL, name: 'general', is_archived: false } + const root = () => ({ type: 'message', ts: ROOT, thread_ts: ROOT, text: '', reply_count: 1 }) + + async function providerFetch(input: string | URL | Request, init?: RequestInit) { + const url = new URL( + typeof input === 'string' ? input : input instanceof URL ? input.href : input.url + ) + expect(url.origin).toBe('https://slack.com') + expect(new Headers(init?.headers).get('Authorization')).toBe('Bearer fixture-slack-token') + const method = url.pathname.split('/').at(-1) + switch (method) { + case 'auth.test': + return Response.json({ ok: true, team_id: TEAM }) + case 'conversations.list': + return Response.json({ ok: true, channels: [channel] }) + case 'conversations.info': + return Response.json({ ok: true, channel }) + case 'conversations.history': + return Response.json({ ok: true, messages: [root()] }) + case 'conversations.replies': + if (missingRoot) return Response.json({ ok: true, messages: [] }) + if (!url.searchParams.has('cursor')) { + return Response.json({ + ok: true, + messages: [root()], + has_more: true, + response_metadata: { next_cursor: 'reply' }, + }) + } + return Response.json({ + ok: true, + messages: [{ type: 'message', ts: REPLY, thread_ts: ROOT, text: replyText }], + is_limited: incomplete, + }) + case 'chat.getPermalink': + return Response.json({ + ok: true, + permalink: `https://fixture.slack.com/archives/${CHANNEL}/p${ROOT.replace('.', '')}`, + }) + default: + throw new Error('Unexpected fixture Slack endpoint') + } + } + async function sync() { + return executeConnectorSyncJob({ + connectorId: ids.connectorId, + requestId: 'slack-empty-fixture', + fullSync: true, + billingAttribution: billing, + }) + } + async function row() { + const [stored] = await db + .select() + .from(document) + .where(and(eq(document.connectorId, ids.connectorId), eq(document.externalId, EXTERNAL_ID))) + expect(stored?.id).toBe(documentId) + return stored! + } + async function vectors() { + return db + .select({ content: embedding.content }) + .from(embedding) + .where(eq(embedding.documentId, documentId)) + } + async function search() { + const result = await searchKnowledge.execute({ + principal: { kind: 'session', userId: ids.aliceId, sessionId: 'slack-empty-fixture' }, + input: { + workspaceId: ids.workspaceId, + knowledgeBaseIds: [ids.knowledgeBaseId], + query: 'Orion', + searchMode: 'hybrid', + topK: 10, + }, + }) + return result.results.map((result) => result.documentId) + } + beforeAll(async () => { + fixture.storageRoot = mkdtempSync(path.join(tmpdir(), 'sim-slack-empty-')) + await seedKnowledgeAclFixture(ids) + billing = await resolveBillingAttribution({ + actorUserId: ids.aliceId, + workspaceId: ids.workspaceId, + }) + await db + .update(knowledgeConnector) + .set({ + connectorType: 'slack', + sourceConfig: { channel: CHANNEL, maxMessages: 0 }, + accessMode: 'workspace', + status: 'active', + syncLockToken: null, + }) + .where(eq(knowledgeConnector.id, ids.connectorId)) + vi.spyOn(connectorTokens, 'resolveConnectorAccessToken').mockResolvedValue({ + accessToken: 'fixture-slack-token', + }) + vi.stubGlobal('fetch', providerFetch) + await db.insert(document).values({ + id: documentId, + knowledgeBaseId: ids.knowledgeBaseId, + connectorId: ids.connectorId, + externalId: EXTERNAL_ID, + filename: 'Thread.txt', + mimeType: 'text/plain', + fileUrl: '', + fileSize: 0, + processingStatus: 'failed', + processingError: 'Synthetic previous source failure', + }) + }) + afterAll(async () => { + await db.delete(workspace).where(eq(workspace.id, ids.workspaceId)) + await db.delete(organization).where(eq(organization.id, ids.organizationId)) + await db.delete(user).where(inArray(user.id, [ids.aliceId, ids.bobId])) + await rm(fixture.storageRoot, { recursive: true, force: true }) + vi.restoreAllMocks() + vi.unstubAllGlobals() + await db.$client.end() + }) + + it('completes empty-thread syncs, recovers reply edits and removes stale searchable text', async () => { + expect(await sync()).toMatchObject({ + outcome: 'completed', + docsFailed: 0, + docsSkipped: 1, + processingDispatch: { requested: 0, failed: 0 }, + }) + expect(await row()).toMatchObject({ + processingStatus: 'failed', + processingError: EMPTY_REASON, + storageKey: null, + }) + expect(await vectors()).toEqual([]) + expect(fixture.embeddingCalls).toBe(0) + expect(await sync()).toMatchObject({ outcome: 'completed', docsFailed: 0 }) + expect(fixture.embeddingCalls).toBe(0) + + replyText = 'Orion launch is scheduled for Friday.' + expect(await sync()).toMatchObject({ outcome: 'completed', docsFailed: 0, docsUpdated: 1 }) + expect(await row()).toMatchObject({ processingStatus: 'completed', processingError: null }) + expect((await vectors()).map((row) => row.content).join(' ')).toContain(replyText) + expect(await search()).toContain(documentId) + const embedded = fixture.embeddingCalls + expect(await sync()).toMatchObject({ outcome: 'completed', docsUnchanged: 1 }) + expect(fixture.embeddingCalls).toBe(embedded) + + replyText = '' + expect(await sync()).toMatchObject({ outcome: 'completed', docsFailed: 0, docsSkipped: 1 }) + expect(await row()).toMatchObject({ processingError: EMPTY_REASON, storageKey: null }) + expect(await vectors()).toEqual([]) + expect(fixture.embeddingCalls).toBe(embedded) + expect(await search()).not.toContain(documentId) + + replyText = 'Orion launch moved to Monday.' + expect(await sync()).toMatchObject({ outcome: 'completed', docsUpdated: 1 }) + expect(await search()).toContain(documentId) + const restored = await vectors() + incomplete = true + replyText = '' + await expect(sync()).rejects.toThrow('1 source failures') + expect(await vectors()).toEqual(restored) + expect((await row()).storageKey).not.toBeNull() + incomplete = false + missingRoot = true + await expect(sync()).rejects.toThrow('1 source failures') + expect(await vectors()).toEqual(restored) + }, 60000) +})