diff --git a/apps/sim/lib/sim-search/live/README.md b/apps/sim/lib/sim-search/live/README.md index 6df8082a441..5aecb3ee723 100644 --- a/apps/sim/lib/sim-search/live/README.md +++ b/apps/sim/lib/sim-search/live/README.md @@ -58,7 +58,7 @@ Each event must exist under the source's delegated token, be in an allowed calen ### Slack -Member mode only. Search uses the connected user's Slack real-time search grant and provider search syntax. Date bounds are sent as numeric provider timestamps; date ordering selects keyword retrieval and timestamp sort, because Slack semantic retrieval uses relevance ordering. Channels, messages, direct messages, and files are limited by that grant and provider scopes; there is no admin channel filter in member mode. Missing required search scopes produce a reconnect status. Thread/file reads use the same connected identity. +Member mode only. Search uses the connected user's Slack real-time search grant and provider search syntax. Date bounds are sent as numeric provider timestamps; date ordering selects keyword retrieval and timestamp sort, because Slack semantic retrieval uses relevance ordering. Channels, messages, direct messages, and files are limited by that grant and provider scopes; there is no admin channel filter in member mode. Missing required search scopes produce a reconnect status. Thread/file reads use the same connected identity. Real-time Search returns at most 20 results a call and counts every page against the member's rate limit, so a search across every source reads one page; a Slack-scoped or native Slack search follows the cursor for up to three pages to fill its limit, and a later page that fails keeps the earlier pages and returns the unread cursor. ### GitHub diff --git a/apps/sim/lib/sim-search/live/providers.test.ts b/apps/sim/lib/sim-search/live/providers.test.ts index a59aeece2c2..e6299d5aa30 100644 --- a/apps/sim/lib/sim-search/live/providers.test.ts +++ b/apps/sim/lib/sim-search/live/providers.test.ts @@ -188,3 +188,199 @@ describe('native search endpoints', () => { expect(api.json).not.toHaveBeenCalled() }) }) + +describe('Slack RTS pages', () => { + const scopes = ['search:read.public'] + const slackPage = (first: number, count: number, nextCursor?: string) => ({ + ok: true, + results: { + messages: Array.from({ length: count }, (_, index) => ({ + message_ts: `${first + index}.000100`, + channel_id: 'C1', + channel_name: 'team', + content: 'Launch', + permalink: `https://team.slack.com/archives/C1/p${first + index}000100`, + })), + }, + response_metadata: { next_cursor: nextCursor ?? '' }, + }) + const sentBodies = (api: ReturnType) => + api.json.mock.calls.map(([, options]) => options?.body as Record) + + it('follows the cursor until a native Slack query fills its limit', async () => { + const api = client() + api.json + .mockResolvedValueOnce(slackPage(1, 20, 'page-2')) + .mockResolvedValueOnce(slackPage(21, 20, 'page-3')) + const result = await searchSlack(api, { + ...input, + limit: 30, + scopes, + native: { provider: 'slack', query: 'launch' }, + }) + expect(sentBodies(api).map(({ cursor, limit }) => ({ cursor, limit }))).toEqual([ + { cursor: undefined, limit: 20 }, + { cursor: 'page-2', limit: 10 }, + ]) + expect(result.documents).toHaveLength(40) + expect(result.nextCursor).toBe('page-3') + expect(result.partial).toBeUndefined() + }) + + it('pages a plain query scoped to the Slack source', async () => { + const api = client() + api.json + .mockResolvedValueOnce(slackPage(1, 20, 'page-2')) + .mockResolvedValueOnce(slackPage(21, 5)) + const result = await searchSlack(api, { + ...input, + limit: 50, + scopes, + filters: { source: 'slack' }, + }) + expect(api.json).toHaveBeenCalledTimes(2) + expect(result.documents).toHaveLength(25) + expect(result.nextCursor).toBeUndefined() + }) + + it('reads one page for a search across every source, whatever its limit', async () => { + const api = client() + api.json.mockResolvedValue(slackPage(1, 20, 'page-2')) + const result = await searchSlack(api, { ...input, limit: 30, scopes }) + expect(api.json).toHaveBeenCalledTimes(1) + expect(result.nextCursor).toBe('page-2') + }) + + it('stops when Slack reports no further page', async () => { + const api = client() + api.json.mockResolvedValue(slackPage(1, 12)) + await searchSlack(api, { + ...input, + limit: 50, + scopes, + native: { provider: 'slack', query: 'launch' }, + }) + expect(api.json).toHaveBeenCalledTimes(1) + }) + + it('reads at most three pages', async () => { + const api = client() + api.json.mockImplementation(async (_path, options) => { + const cursor = (options?.body as { cursor?: string }).cursor + const index = cursor ? Number(cursor) : 0 + return slackPage(index * 20 + 1, 1, String(index + 1)) + }) + const result = await searchSlack(api, { + ...input, + limit: 100, + scopes, + native: { provider: 'slack', query: 'launch' }, + }) + expect(api.json).toHaveBeenCalledTimes(3) + expect(result.nextCursor).toBe('3') + }) + + it('stops once matched files fill the limit alongside messages', async () => { + const api = client() + const withFiles = slackPage(1, 20, 'page-2') + api.json.mockResolvedValue({ + ...withFiles, + results: { + ...withFiles.results, + files: Array.from({ length: 10 }, (_, index) => ({ + file_id: `F${index}`, + title: 'Launch plan', + permalink: `https://team.slack.com/files/F${index}`, + })), + }, + }) + const result = await searchSlack(api, { + ...input, + limit: 30, + scopes: [...scopes, 'search:read.files'], + native: { provider: 'slack', query: 'launch' }, + }) + expect(api.json).toHaveBeenCalledTimes(1) + expect(result.documents).toHaveLength(30) + }) + + it('keeps earlier pages and the unread cursor when a later page fails', async () => { + const api = client() + api.json + .mockResolvedValueOnce(slackPage(1, 20, 'page-2')) + .mockResolvedValueOnce({ ok: false, error: 'ratelimited' }) + const result = await searchSlack(api, { + ...input, + limit: 50, + scopes, + native: { provider: 'slack', query: 'launch' }, + }) + expect(result.documents).toHaveLength(20) + expect(result.nextCursor).toBe('page-2') + expect(result.partial).toBe(true) + expect(result.message).toContain('nextCursor') + }) + + it('keeps earlier pages when a later page fails outside the Slack API', async () => { + const api = client() + api.json + .mockResolvedValueOnce(slackPage(1, 20, 'page-2')) + .mockRejectedValueOnce(new Error('socket hang up')) + const result = await searchSlack(api, { + ...input, + limit: 50, + scopes, + native: { provider: 'slack', query: 'launch' }, + }) + expect(result.documents).toHaveLength(20) + expect(result.nextCursor).toBe('page-2') + expect(result.partial).toBe(true) + }) + + it('stops paging once the caller cancels the search', async () => { + const api = client() + const cancelled = new AbortController() + api.json.mockResolvedValueOnce(slackPage(1, 20, 'page-2')).mockImplementationOnce(async () => { + cancelled.abort() + throw new Error('aborted') + }) + await expect( + searchSlack(api, { + ...input, + limit: 50, + scopes, + signal: cancelled.signal, + native: { provider: 'slack', query: 'launch' }, + }) + ).rejects.toThrow('aborted') + }) + + it('keeps paging past short pages that still carry a cursor', async () => { + const api = client() + api.json + .mockResolvedValueOnce(slackPage(1, 10, 'page-2')) + .mockResolvedValueOnce(slackPage(11, 10, 'page-3')) + .mockResolvedValueOnce(slackPage(21, 10)) + const result = await searchSlack(api, { + ...input, + limit: 30, + scopes, + native: { provider: 'slack', query: 'launch' }, + }) + expect(api.json).toHaveBeenCalledTimes(3) + expect(result.documents).toHaveLength(30) + }) + + it('still fails when the first page fails', async () => { + const api = client() + api.json.mockResolvedValue({ ok: false, error: 'ratelimited' }) + await expect( + searchSlack(api, { + ...input, + limit: 50, + scopes, + native: { provider: 'slack', query: 'launch' }, + }) + ).rejects.toMatchObject({ status: 'rate_limited' }) + }) +}) diff --git a/apps/sim/lib/sim-search/live/slack.ts b/apps/sim/lib/sim-search/live/slack.ts index eb08c0e2a03..39701e4c970 100644 --- a/apps/sim/lib/sim-search/live/slack.ts +++ b/apps/sim/lib/sim-search/live/slack.ts @@ -1,5 +1,6 @@ import { dateSortDirection, nativeDateBounds, nativeText } from '@/lib/sim-search/live/dates' import { array, NativeSearchError, object, string } from '@/lib/sim-search/live/http' +import { joinMessages } from '@/lib/sim-search/live/pages' import { slackConversationName, slackConversationUrl, @@ -35,6 +36,20 @@ function slackResult(value: unknown) { return data } +/** Results one Real-time Search call returns at most. */ +const SLACK_PAGE_SIZE = 20 +/** + * Real-time Search pages a search aimed at Slack reads to fill its limit. Each page counts + * against the member's RTS rate limit of about 10 calls a minute. + */ +const MAX_SLACK_SEARCH_PAGES = 3 + +/** + * A search across every source reads one page, so Slack shares the result list with the other + * sources. A Slack-scoped or native Slack search follows the cursor until it fills its limit. + * A later page that fails keeps the earlier pages and returns the cursor it could not read, + * unless the caller cancelled the search. + */ export async function searchSlack( client: NativeClient, input: NativeSearchInput @@ -44,6 +59,40 @@ export async function searchSlack( 'reconnect', 'Reconnect your Slack account to grant permission to search.' ) + const pages = input.native || input.filters?.source === 'slack' ? MAX_SLACK_SEARCH_PAGES : 1 + const first = await searchSlackPage(client, input, input.native?.cursor, input.limit) + const documents = [...first.documents] + let nextCursor = first.nextCursor + for (let page = 1; page < pages && nextCursor && documents.length < input.limit; page++) { + try { + const more = await searchSlackPage(client, input, nextCursor, input.limit - documents.length) + documents.push(...more.documents) + nextCursor = more.nextCursor + } catch (error) { + if (input.signal?.aborted) throw error + return { + documents, + nextCursor, + partial: true, + message: joinMessages([ + first.message, + error instanceof NativeSearchError + ? error.message + : 'Slack could not return more results.', + 'Continue with nextCursor for more.', + ]), + } + } + } + return { documents, nextCursor, message: first.message } +} + +async function searchSlackPage( + client: NativeClient, + input: NativeSearchInput, + cursor: string | undefined, + limit: number +): Promise { const channels = [ ['public_channel', 'search:read.public'], ['private_channel', 'search:read.private'], @@ -80,13 +129,13 @@ export async function searchSlack( ? ['messages', 'files'] : ['messages'], include_context_messages: true, - limit: Math.min(input.limit, 20), + limit: Math.min(limit, SLACK_PAGE_SIZE), ...(dates.start ? { after: Math.floor(Date.parse(dates.start) / 1000) - 1 } : {}), ...(dates.end ? { before: Math.ceil(Date.parse(dates.end) / 1000) } : {}), ...(dateSortDirection(input.filters) ? { sort: 'timestamp', sort_dir: dateSortDirection(input.filters) } : {}), - ...(input.native?.cursor ? { cursor: input.native.cursor } : {}), + ...(cursor ? { cursor } : {}), ...(input.native?.termClauses ? { term_clauses: input.native.termClauses } : {}), ...(modifiers ? { modifiers } : {}), ...(input.native?.keywordOnly || !nativeText(input)