fix: add support for new list request-queue requests parameters (#880)

add support for endpoint parameters and new pagination method added in
apify/apify-core#12115

There is a new parameter `filter` for the _request queue -> list
requests_ endpoint. To properly support pagination, we also added proper
cursor-based pagination, with opaque cursor provided by the server. So
this PR switches the pagination internals to use this new way. All of
this is done in backwards-compatible manner, the test updates are only
necessary to update the server implementation mocks and cases where the
existing tests were inspecting some internals.

Author: mvolfik

src/resource_clients/request_queue.ts

@@ -12,9 +12,10 @@ import type { ApifyRequestConfig } from '../http_client';
 import {
     cast,
     catchNotFoundOrThrow,
-    PaginationIterator,
+    mutuallyExclusive,
     parseDateFields,
     pluckData,
+    RequestQueuePaginationIterator,
     sliceArrayByByteLength,
 } from '../utils';
 
@@ -682,10 +683,14 @@ export class RequestQueueClient extends ResourceClient {
     ): Promise<RequestQueueClientListRequestsResult> & AsyncIterable<RequestQueueClientRequestSchema> {
         ow(
             options,
-            ow.object.exactShape({
-                limit: ow.optional.number.not.negative,
-                exclusiveStartId: ow.optional.string,
-            }),
+            ow.object
+                .exactShape({
+                    limit: ow.optional.number.not.negative,
+                    exclusiveStartId: ow.optional.string,
+                    cursor: ow.optional.string,
+                    filter: ow.optional.array.ofType(ow.string.oneOf(['locked', 'pending'])).minLength(1),
+                })
+                .validate(mutuallyExclusive('exclusiveStartId', 'cursor')),
         );
 
         const getPaginatedList = async (
@@ -695,7 +700,11 @@ export class RequestQueueClient extends ResourceClient {
                 url: this._url('requests'),
                 method: 'GET',
                 timeout: Math.min(MEDIUM_TIMEOUT_MILLIS, this.timeoutMillis ?? Infinity),
-                params: this._params({ ...rqListOptions, clientKey: this.clientKey }),
+                params: this._params({
+                    ...rqListOptions,
+                    filter: rqListOptions.filter ? rqListOptions.filter.join(',') : undefined,
+                    clientKey: this.clientKey,
+                }),
             });
 
             return cast(parseDateFields(pluckData(response.data)));
@@ -712,10 +721,16 @@ export class RequestQueueClient extends ResourceClient {
             // of exhausting all requests we get response with empty items which ends the loop.
             while (
                 currentPage.items.length > 0 && // Continue only if at least some items were returned in the last page.
+                currentPage.nextCursor && // Continue only if the API returned a cursor for the next page.
                 (remainingItems === undefined || remainingItems > 0) // Continue only if the limit was not exceeded.
             ) {
-                const exclusiveStartId = currentPage.items[currentPage.items.length - 1].id;
-                const newOptions = { ...options, limit: remainingItems, exclusiveStartId };
+                const newOptions = {
+                    ...options,
+                    limit: remainingItems,
+                    // remove original exclusiveStartId, if there was any, and use cursor-based pagination
+                    exclusiveStartId: undefined,
+                    cursor: currentPage.nextCursor,
+                };
                 currentPage = await getPaginatedList(newOptions);
                 yield* currentPage.items;
                 if (remainingItems) {
@@ -773,17 +788,29 @@ export class RequestQueueClient extends ResourceClient {
     ): RequestQueueRequestsAsyncIterable<RequestQueueClientListRequestsResult> {
         ow(
             options,
-            ow.object.exactShape({
-                limit: ow.optional.number.not.negative,
-                maxPageLimit: ow.optional.number,
-                exclusiveStartId: ow.optional.string,
-            }),
+            ow.object
+                .exactShape({
+                    limit: ow.optional.number.not.negative,
+                    maxPageLimit: ow.optional.number,
+                    exclusiveStartId: ow.optional.string,
+                    cursor: ow.optional.string,
+                    filter: ow.optional.array.ofType(ow.string.oneOf(['locked', 'pending'])).minLength(1),
+                })
+                .validate(mutuallyExclusive('exclusiveStartId', 'cursor')),
         );
-        const { limit, exclusiveStartId, maxPageLimit = DEFAULT_REQUEST_QUEUE_REQUEST_PAGE_LIMIT } = options;
-        return new PaginationIterator({
-            getPage: this.listRequests.bind(this),
+
+        const {
+            limit,
+            exclusiveStartId,
+            cursor,
+            filter,
+            maxPageLimit = DEFAULT_REQUEST_QUEUE_REQUEST_PAGE_LIMIT,
+        } = options;
+        return new RequestQueuePaginationIterator({
+            getPage: async (pageOptions) => this.listRequests({ ...pageOptions, filter }),
             limit,
             exclusiveStartId,
+            cursor,
             maxPageLimit,
         });
     }
@@ -860,13 +887,20 @@ export interface RequestQueueClientListHeadResult {
     items: RequestQueueClientListItem[];
 }
 
+export type RequestQueueListRequestsFilter = 'locked' | 'pending';
+
 /**
  * Options for listing all requests in the queue.
  */
 export interface RequestQueueClientListRequestsOptions {
     limit?: number;
-    /* Using id of request that does not exist in request queue leads to unpredictable results. */
+    /**
+     * Using id of request that does not exist in request queue leads to unpredictable results.
+     * @deprecated Use `cursor` for pagination instead.
+     */
     exclusiveStartId?: string;
+    cursor?: string;
+    filter?: readonly RequestQueueListRequestsFilter[];
 }
 
 /**
@@ -875,15 +909,21 @@ export interface RequestQueueClientListRequestsOptions {
 export interface RequestQueueClientPaginateRequestsOptions {
     limit?: number;
     maxPageLimit?: number;
+    /** @deprecated Use `cursor` for pagination instead. */
     exclusiveStartId?: string;
+    cursor?: string;
+    filter?: readonly RequestQueueListRequestsFilter[];
 }
 
 /**
  * Result of listing all requests in the queue.
  */
 export interface RequestQueueClientListRequestsResult {
     limit: number;
+    /** @deprecated Use `cursor` for pagination instead. */
     exclusiveStartId?: string;
+    cursor?: string;
+    nextCursor?: string;
     items: RequestQueueClientRequestSchema[];
 }
 

src/utils.ts

@@ -186,7 +186,7 @@ export function getVersionData(): { version: string } {
 /**
  * Helper class to create async iterators from paginated list endpoints with exclusive start key.
  */
-export class PaginationIterator {
+export class RequestQueuePaginationIterator {
     private readonly maxPageLimit: number;
 
     private readonly getPage: (
@@ -196,33 +196,41 @@ export class PaginationIterator {
     private readonly limit?: number;
 
     private readonly exclusiveStartId?: string;
+    private readonly cursor?: string;
 
-    constructor(options: PaginationIteratorOptions) {
+    constructor(options: RequestQueuePaginationIteratorOptions) {
         this.maxPageLimit = options.maxPageLimit;
         this.limit = options.limit;
         this.exclusiveStartId = options.exclusiveStartId;
+        this.cursor = options.cursor;
         this.getPage = options.getPage;
     }
 
     async *[Symbol.asyncIterator](): AsyncIterator<RequestQueueClientListRequestsResult> {
-        let nextPageExclusiveStartId;
+        let nextCursor = this.cursor;
+        // allow using exclusiveStartId for the first page, but then we'll delete it to avoid using it for any later page
+        let nextExclusiveStartId = this.exclusiveStartId;
+
         let iterateItemCount = 0;
         while (true) {
             const pageLimit = this.limit
                 ? Math.min(this.maxPageLimit, this.limit - iterateItemCount)
                 : this.maxPageLimit;
-            const pageExclusiveStartId = nextPageExclusiveStartId || this.exclusiveStartId;
+
             const page: RequestQueueClientListRequestsResult = await this.getPage({
                 limit: pageLimit,
-                exclusiveStartId: pageExclusiveStartId,
+                cursor: nextCursor,
+                exclusiveStartId: nextExclusiveStartId,
             });
             // There are no more pages to iterate
             if (page.items.length === 0) return;
             yield page;
             iterateItemCount += page.items.length;
             // Limit reached stopping to iterate
-            if (this.limit && iterateItemCount >= this.limit) return;
-            nextPageExclusiveStartId = page.items[page.items.length - 1].id;
+            if ((this.limit && iterateItemCount >= this.limit) || !page.nextCursor) return;
+
+            nextCursor = page.nextCursor;
+            nextExclusiveStartId = undefined; // see comment above - delete it for any page after the first one, and paginate with cursor
         }
     }
 }
@@ -235,11 +243,12 @@ declare global {
 /**
  * Options for creating a pagination iterator.
  */
-export interface PaginationIteratorOptions {
+export interface RequestQueuePaginationIteratorOptions {
     maxPageLimit: number;
     getPage: (opts: RequestQueueClientListRequestsOptions) => Promise<RequestQueueClientListRequestsResult>;
     limit?: number;
     exclusiveStartId?: string;
+    cursor?: string;
 }
 
 /**
@@ -345,3 +354,13 @@ export function applyQueryParamsToUrl(
     }
     return url;
 }
+
+export const mutuallyExclusive =
+    <T, K extends keyof T>(...keys: K[]) =>
+    (value: T) => {
+        const presentKeys = keys.filter((key) => typeof value[key] !== 'undefined');
+        return {
+            validator: presentKeys.length <= 1,
+            message: `At most one of the following fields is allowed: ${keys.join(', ')}`,
+        };
+    };

test/pagination.test.ts

@@ -359,6 +359,11 @@ describe('RequestQueueClient.listKeys as async iterable', () => {
             userDefinedOptions: { exclusiveStartId: '1000' },
             expectedItems: range(1001, 2500),
         },
+        {
+            testName: 'cursor',
+            userDefinedOptions: { cursor: 'cursor:1000' },
+            expectedItems: range(1000, 2500),
+        },
     ];
 
     const testCases = generateTestCases(
@@ -376,17 +381,23 @@ describe('RequestQueueClient.listKeys as async iterable', () => {
             // There are 2500 items in the collection.
             // Items are simple objects with incrementing attributes for easy verification.
 
-            const exclusiveStartId = request.params.exclusiveStartId ? request.params.exclusiveStartId : null;
             const limit = request.params.limit ? request.params.limit : 0;
 
             if (limit < 0) {
                 throw new Error('Limit must be non-negative');
             }
 
-            const lowerIndex = Math.min(
-                request.params.exclusiveStartId ? Number(request.params.exclusiveStartId) + 1 : 0,
-                totalItems,
-            );
+            if (request.params.exclusiveStartId && request.params.cursor) {
+                throw new Error('exclusiveStartId and cursor cannot be used together');
+            }
+            let effectiveStartIndex = 0;
+            if (request.params.exclusiveStartId) {
+                effectiveStartIndex = Number(request.params.exclusiveStartId) + 1;
+            } else if (request.params.cursor) {
+                effectiveStartIndex = Number(request.params.cursor.split(':')[1]);
+            }
+
+            const lowerIndex = Math.min(effectiveStartIndex, totalItems);
             const upperIndex = Math.min(lowerIndex + (limit || totalItems), totalItems, lowerIndex + maxItemsPerPage);
             const items = range(lowerIndex, upperIndex);
 
@@ -396,7 +407,9 @@ describe('RequestQueueClient.listKeys as async iterable', () => {
                         items,
                         count: items.length,
                         limit: limit || maxItemsPerPage,
-                        exclusiveStartId,
+                        exclusiveStartId: request.params.exclusiveStartId,
+                        cursor: request.params.cursor,
+                        nextCursor: items.length < maxItemsPerPage ? undefined : `cursor:${upperIndex}`,
                     },
                 },
             };
@@ -410,11 +423,7 @@ describe('RequestQueueClient.listKeys as async iterable', () => {
                 items.push(page);
             }
 
-            let expectedAPIcalls = Math.max(Math.ceil(expectedItems.length / maxItemsPerPage), 1);
-            if (userDefinedOptions.limit === undefined || userDefinedOptions.limit > totalItems) {
-                // One extra call to confirm there are no more items due RQ API design.
-                expectedAPIcalls += 1;
-            }
+            const expectedAPIcalls = Math.max(Math.ceil(expectedItems.length / maxItemsPerPage), 1);
 
             expect(items).toEqual(expectedItems);
             expect(mockedClient).toHaveBeenCalledTimes(expectedAPIcalls);

test/request_queues.test.ts

@@ -1,6 +1,6 @@
 import type { AddressInfo } from 'node:net';
 
-import type { Dictionary } from 'apify-client';
+import type { Dictionary, RequestQueueClientListRequestsOptions } from 'apify-client';
 import { ApifyClient } from 'apify-client';
 import type { Request } from 'express';
 import type { Page } from 'puppeteer';
@@ -525,20 +525,29 @@ describe('Request Queue methods', () => {
             validateRequest({ query: {}, params: { queueId, requestId } });
         });
 
-        test('listRequests() works', async () => {
+        test.each([
+            [undefined, undefined],
+            [['pending'], 'pending'],
+            [['pending', 'locked'], 'pending,locked'],
+        ] as const)('listRequests() works', async (filter, filterInQuery) => {
             const queueId = 'some-id';
-            const options = { limit: 5, exclusiveStartId: '123' };
+            const options = { limit: 5, exclusiveStartId: '123' } as RequestQueueClientListRequestsOptions;
+            const queryForValidation = { limit: '5', exclusiveStartId: '123' } as Record<string, string>;
+            if (filter) {
+                options.filter = filter;
+                queryForValidation.filter = filterInQuery;
+            }
 
             const res = await client.requestQueue(queueId).listRequests(options);
-            validateRequest({ query: options, params: { queueId }, endpointId: 'list-requests' });
+            validateRequest({ query: queryForValidation, params: { queueId }, endpointId: 'list-requests' });
 
             const browserRes = await page.evaluate(
                 (id, opts) => client.requestQueue(id).listRequests(opts),
                 queueId,
                 options,
             );
             expect(browserRes).toEqual(res);
-            validateRequest({ query: options, params: { queueId } });
+            validateRequest({ query: queryForValidation, params: { queueId } });
         });
 
         test('paginateRequests() works', async () => {
@@ -551,6 +560,8 @@ describe('Request Queue methods', () => {
                     body: {
                         data: {
                             items,
+                            nextCursor:
+                                items.length > 0 ? `the-request-after-${items[items.length - 1].id}` : undefined,
                         },
                     },
                 });
@@ -559,17 +570,17 @@ describe('Request Queue methods', () => {
             const pagination = client.requestQueue(queueId).paginateRequests({ maxPageLimit });
             // Mock API call for the first page
             let expectedItemsInPage = requests.splice(0, maxPageLimit);
-            let expectedExclusiveStartId;
+            let expectedCursor;
             mockResponse(expectedItemsInPage);
             for await (const { items } of pagination) {
                 expect(items).toEqual(expectedItemsInPage);
                 // Validate the request for the current iteration page
                 validateRequest({
-                    query: { exclusiveStartId: expectedExclusiveStartId, limit: maxPageLimit },
+                    query: { cursor: expectedCursor, limit: maxPageLimit },
                     params: { queueId },
                 });
                 // Prepare expectations and mock request for the next iteration page
-                expectedExclusiveStartId = items[items.length - 1].id;
+                expectedCursor = `the-request-after-${items[items.length - 1].id}`;
                 expectedItemsInPage = requests.splice(0, maxPageLimit);
                 mockResponse(expectedItemsInPage);
             }