improvement(search): page Slack Real-time Search to fill a Slack-scoped search's limit

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
 

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<typeof client>) =>
+    api.json.mock.calls.map(([, options]) => options?.body as Record<string, unknown>)
+
+  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' })
+  })
+})

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<NativePage> {
   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)