fix: paginate MCP inbox items (#1329)

* fix: paginate MCP inbox items * fix: validate MCP inbox cursors * test: reset MCP inbox pagination state
2026-08-01 18:50:42 +02:00
parent bcf919d995
commit bfbd926950
2 changed files with 223 additions and 27 deletions
@@ -1,9 +1,51 @@
-import { describe, expect, it } from 'vitest'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { eventBus } from '@/lib/events/bus'
 import { createQueuedMockSupabase } from '@/tests/helpers'
 import { tools } from '../server'

 const tool = tools.find((candidate) => candidate.name === 'gnubok_list_inbox_items')!

+beforeEach(() => {
+  vi.clearAllMocks()
+  eventBus.clear()
+})
+
+function makeInboxItem(overrides: Record<string, unknown> = {}) {
+  return {
+    id: '11111111-1111-4111-8111-111111111111',
+    status: 'received',
+    source: 'upload',
+    created_at: '2026-07-01T12:00:00Z',
+    extracted_data: null,
+    matched_supplier_id: null,
+    matched_transaction_id: null,
+    created_supplier_invoice_id: null,
+    created_journal_entry_id: null,
+    email_from: null,
+    email_subject: null,
+    error_message: null,
+    document_attachments: [],
+    ...overrides,
+  }
+}
+
+function makeRecordingChain(result: { data: unknown; error: unknown }) {
+  const calls: Array<{ method: string; args: unknown[] }> = []
+  const handler: ProxyHandler<object> = {
+    get(_target, prop) {
+      if (prop === 'then') {
+        return (resolve: (value: unknown) => void) => resolve(result)
+      }
+      return (...args: unknown[]) => {
+        calls.push({ method: String(prop), args })
+        return proxy
+      }
+    },
+  }
+  const proxy = new Proxy({}, handler)
+  return { proxy, calls }
+}
+
 describe('gnubok_list_inbox_items', () => {
  it('advertises file_name in its item output contract', () => {
    const schema = tool.outputSchema as {
@@ -24,40 +66,32 @@ describe('gnubok_list_inbox_items', () => {
    expect(schema.properties.items.items.required).toContain('file_name')
  })

+  it('advertises optional cursor pagination without changing required output fields', () => {
+    const inputSchema = tool.inputSchema as { properties: Record<string, unknown> }
+    const outputSchema = tool.outputSchema as {
+      properties: Record<string, unknown>
+      required: string[]
+    }
+
+    expect(inputSchema.properties.cursor).toBeDefined()
+    expect(outputSchema.properties.next_cursor).toBeDefined()
+    expect(outputSchema.required).toEqual(['items', 'count'])
+  })
+
  it('joins the document and returns its file_name on each list row', async () => {
    const { supabase, enqueue, findCall } = createQueuedMockSupabase()
    enqueue({
      data: [
-        {
+        makeInboxItem({
          id: 'inbox-1',
-          status: 'received',
-          source: 'upload',
          created_at: '2026-07-31T12:00:00Z',
-          extracted_data: null,
-          matched_supplier_id: null,
-          matched_transaction_id: null,
-          created_supplier_invoice_id: null,
-          created_journal_entry_id: null,
-          email_from: null,
-          email_subject: null,
-          error_message: null,
          document_attachments: [{ file_name: 'dooer-export-2026-07.pdf' }],
-        },
-        {
+        }),
+        makeInboxItem({
          id: 'inbox-2',
-          status: 'received',
          source: 'email',
          created_at: '2026-07-30T12:00:00Z',
-          extracted_data: null,
-          matched_supplier_id: null,
-          matched_transaction_id: null,
-          created_supplier_invoice_id: null,
-          created_journal_entry_id: null,
-          email_from: null,
-          email_subject: null,
-          error_message: null,
-          document_attachments: [],
-        },
+        }),
      ],
      error: null,
    })
@@ -75,4 +109,112 @@ describe('gnubok_list_inbox_items', () => {
    ])
    expect(result.count).toBe(2)
  })
+
+  it('preserves the first-page response when there are no more items', async () => {
+    const { supabase, enqueue } = createQueuedMockSupabase()
+    enqueue({ data: [makeInboxItem()], error: null })
+
+    const result = await tool.execute({}, 'company-1', 'user-1', supabase as never)
+
+    expect(result).toMatchObject({
+      count: 1,
+      items: [{ id: '11111111-1111-4111-8111-111111111111', created_at: '2026-07-01T12:00:00Z' }],
+    })
+    expect(result).not.toHaveProperty('next_cursor')
+  })
+
+  it('returns a composite cursor for a full page', async () => {
+    const { supabase, enqueue } = createQueuedMockSupabase()
+    enqueue({
+      data: [
+        makeInboxItem({ id: '22222222-2222-4222-8222-222222222222', created_at: '2026-07-02T12:00:00Z' }),
+        makeInboxItem({ id: '11111111-1111-4111-8111-111111111111', created_at: '2026-07-01T12:00:00Z' }),
+      ],
+      error: null,
+    })
+
+    const result = await tool.execute({ limit: 2 }, 'company-1', 'user-1', supabase as never)
+
+    expect(result).toMatchObject({
+      count: 2,
+      next_cursor: '2026-07-01T12:00:00Z__11111111-1111-4111-8111-111111111111',
+    })
+  })
+
+  it('uses stable keyset ordering and applies a composite cursor exclusively', async () => {
+    const query = makeRecordingChain({ data: [], error: null })
+    const supabase = { from: vi.fn().mockReturnValue(query.proxy) }
+
+    await tool.execute(
+      { cursor: '2026-07-01T12:00:00Z__11111111-1111-4111-8111-111111111111' },
+      'company-1',
+      'user-1',
+      supabase as never,
+    )
+
+    expect(query.calls.filter((call) => call.method === 'order')).toEqual([
+      { method: 'order', args: ['created_at', { ascending: false }] },
+      { method: 'order', args: ['id', { ascending: false }] },
+    ])
+    expect(query.calls).toContainEqual({
+      method: 'or',
+      args: [
+        'created_at.lt.2026-07-01T12:00:00Z,and(created_at.eq.2026-07-01T12:00:00Z,id.lt.11111111-1111-4111-8111-111111111111)',
+      ],
+    })
+  })
+
+  it.each([
+    'not-a-timestamp',
+    '2026-07-01T12:00:00Z__11111111-1111-4111-8111-111111111111,created_at.gt.1900-01-01',
+  ])('rejects a malformed cursor before querying the database: %s', async (cursor) => {
+    const supabase = { from: vi.fn() }
+
+    await expect(
+      tool.execute({ cursor }, 'company-1', 'user-1', supabase as never),
+    ).rejects.toThrow(/Invalid cursor/)
+    expect(supabase.from).not.toHaveBeenCalled()
+  })
+
+  it('advances a full unprocessed scan window even when every row is processed', async () => {
+    const { supabase, enqueue } = createQueuedMockSupabase()
+    const rows = Array.from({ length: 200 }, (_, index) =>
+      makeInboxItem({
+        id: `00000000-0000-4000-8000-${String(200 - index).padStart(12, '0')}`,
+        created_at: `2026-07-01T11:${String(59 - (index % 60)).padStart(2, '0')}:00Z`,
+        matched_transaction_id: `transaction-${index}`,
+      }),
+    )
+    enqueue({ data: rows, error: null })
+
+    const result = await tool.execute(
+      { unprocessed_only: true },
+      'company-1',
+      'user-1',
+      supabase as never,
+    )
+
+    expect(result).toEqual({
+      items: [],
+      count: 0,
+      next_cursor: `${rows[199].created_at}__${rows[199].id}`,
+    })
+  })
+
+  it('supports the legacy timestamp-only cursor form', async () => {
+    const query = makeRecordingChain({ data: [], error: null })
+    const supabase = { from: vi.fn().mockReturnValue(query.proxy) }
+
+    await tool.execute(
+      { cursor: '2026-07-01T12:00:00Z' },
+      'company-1',
+      'user-1',
+      supabase as never,
+    )
+
+    expect(query.calls).toContainEqual({
+      method: 'lt',
+      args: ['created_at', '2026-07-01T12:00:00Z'],
+    })
+  })
 })
@@ -9167,6 +9167,12 @@ export const tools: McpTool[] = [
        status: { type: 'string', enum: ['received', 'error'], description: 'Filter by status' },
        unprocessed_only: { type: 'boolean', description: 'When true, only return items with no terminal link yet (not matched to a transaction, supplier invoice, or journal entry), i.e. documents that still need handling. Default false.' },
        limit: { type: 'number', description: 'Max results (default 20, max 50)' },
+        cursor: {
+          type: 'string',
+          maxLength: 100,
+          pattern: '^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}(?:\\.\\d+)?(?:Z|[+-]\\d{2}:\\d{2})(?:__[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})?$',
+          description: 'Composite "<created_at>__<inbox_item_id>" from previous page (exclusive). Pass next_cursor verbatim.',
+        },
      },
    },
    outputSchema: {
@@ -9187,6 +9193,7 @@ export const tools: McpTool[] = [
          },
        },
        count: { type: 'number' },
+        next_cursor: { type: 'string', description: 'Pass as cursor on next call. Absent = no more pages.' },
      },
      required: ['items', 'count'],
    },
@@ -9200,6 +9207,29 @@ export const tools: McpTool[] = [
      const limit = Math.min(Math.max(1, Number(args.limit) || 20), 50)
      const status = args.status as string | undefined
      const unprocessedOnly = args.unprocessed_only === true
+      const cursor = typeof args.cursor === 'string' ? args.cursor : null
+
+      // Composite cursor: "<created_at>__<id>". Falls back to plain timestamp
+      // for backward compatibility with older callers.
+      let cursorTs: string | null = null
+      let cursorId: string | null = null
+      if (cursor) {
+        const sep = cursor.indexOf('__')
+        if (sep === -1) {
+          cursorTs = cursor
+        } else {
+          cursorTs = cursor.slice(0, sep)
+          cursorId = cursor.slice(sep + 2)
+        }
+      }
+      if (cursorTs && !z.string().datetime({ offset: true }).safeParse(cursorTs).success) {
+        throw new Error('Invalid cursor timestamp. Pass next_cursor verbatim.')
+      }
+      if (cursorId && !z.string().uuid().safeParse(cursorId).success) {
+        throw new Error('Invalid cursor inbox item ID. Pass next_cursor verbatim.')
+      }
+
+      const fetchSize = unprocessedOnly ? 200 : limit

      let query = supabase
        .from('invoice_inbox_items')
@@ -9210,11 +9240,19 @@ export const tools: McpTool[] = [
        `)
        .eq('company_id', companyId)
        .order('created_at', { ascending: false })
+        .order('id', { ascending: false })
        // Fetch a wider window when filtering client-side so the limit
        // applies to the post-filter set rather than truncating before it.
-        .limit(unprocessedOnly ? 200 : limit)
+        .limit(fetchSize)

      if (status) query = query.eq('status', status)
+      if (cursorTs && cursorId) {
+        query = query.or(
+          `created_at.lt.${cursorTs},and(created_at.eq.${cursorTs},id.lt.${cursorId})`
+        )
+      } else if (cursorTs) {
+        query = query.lt('created_at', cursorTs)
+      }

      const { data, error } = await query
      if (error) throw new Error(`Database error: ${error.message}`)
@@ -9268,7 +9306,23 @@ export const tools: McpTool[] = [
      const filtered = unprocessedOnly ? mapped.filter((i) => !i.processed) : mapped
      const items = filtered.slice(0, limit)

-      return { items, count: items.length }
+      // A full returned page continues after its last item. When client-side
+      // filtering yields a short page from a full scan window, continue after
+      // the last inspected row so older unprocessed items remain reachable.
+      let nextCursor: string | null = null
+      if (items.length === limit) {
+        const last = items[items.length - 1]
+        nextCursor = `${last.created_at}__${last.id}`
+      } else if (data && data.length === fetchSize) {
+        const last = data[data.length - 1]
+        nextCursor = `${last.created_at}__${last.id}`
+      }
+
+      return {
+        items,
+        count: items.length,
+        ...(nextCursor ? { next_cursor: nextCursor } : {}),
+      }
    },
  },