diff --git a/foundations/core/packages/api-client/src/client.ts b/foundations/core/packages/api-client/src/client.ts index ab3eb5a7f8..cc52f7035c 100644 --- a/foundations/core/packages/api-client/src/client.ts +++ b/foundations/core/packages/api-client/src/client.ts @@ -22,10 +22,13 @@ import { type Doc, type DocumentQuery, type FindOptions, + type FindPageOptions, + type FindPageResult, type FindResult, type Hierarchy, type ModelDb, type Ref, + type IterateOptions, type Space, type TxResult, type WithLookup, @@ -156,6 +159,22 @@ class PlatformClientImpl implements PlatformClient { return await this.client.findAll(_class, query, options) } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + return await this.client.findAllPage(_class, query, options) + } + + iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + return this.client.iterateAll(_class, query, options) + } + async close (): Promise { await this.connection.close() } diff --git a/foundations/core/packages/api-client/src/rest/adapter.ts b/foundations/core/packages/api-client/src/rest/adapter.ts index 18f1cf8afe..ea0039e6e4 100644 --- a/foundations/core/packages/api-client/src/rest/adapter.ts +++ b/foundations/core/packages/api-client/src/rest/adapter.ts @@ -22,10 +22,13 @@ import { type DomainRequestOptions, type DomainResult, type FindOptions, + type FindPageOptions, + type FindPageResult, type FindResult, Hierarchy, ModelDb, OperationDomain, + type IterateOptions, type Ref, type SearchOptions, type SearchQuery, @@ -60,6 +63,22 @@ export class RestClientAdapter implements Client { return await this.client.findAll(_class, query, options) } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + return await this.client.findAllPage(_class, query, options) + } + + iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + return this.client.iterateAll(_class, query, options) + } + async tx (tx: Tx): Promise { return await this.client.tx(tx) } diff --git a/foundations/core/packages/api-client/src/rest/rest.ts b/foundations/core/packages/api-client/src/rest/rest.ts index de2dd6ac57..a098ca9703 100644 --- a/foundations/core/packages/api-client/src/rest/rest.ts +++ b/foundations/core/packages/api-client/src/rest/rest.ts @@ -28,7 +28,10 @@ import { type DomainRequestOptions, type DomainResult, type FindOptions, + type FindPageOptions, + type FindPageResult, type FindResult, + type IterateOptions, Hierarchy, MeasureMetricsContext, type Mixin, @@ -165,6 +168,81 @@ export class RestClientImpl implements RestClient { return result } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + const requestUrl = concatLink(this.endpoint, `/api/v1/find-page/${this.workspace}`) + const result = await withRetry & { error?: Status }>(async () => { + const response = await fetch(requestUrl, { + method: 'POST', + keepalive: true, + headers: this.jsonHeaders(), + body: JSON.stringify({ _class, query, options }) + }) + if (!response.ok) { + await this.checkRateLimits(response) + throw new PlatformError(unknownError(response.statusText)) + } + this.updateRateLimit(response) + return await extractJson>(response) + }, isRLE) + + if (result.error !== undefined) { + throw new PlatformError(result.error) + } + if (result.lookupMap !== undefined) { + for (const doc of result.docs) { + if (doc.$lookup !== undefined) { + const lookup = doc.$lookup as Record + for (const [key, value] of Object.entries(lookup)) { + if (Array.isArray(value)) { + lookup[key] = value.map((item) => result.lookupMap?.[item]) + } else { + lookup[key] = result.lookupMap[value as string] + } + } + } + } + delete result.lookupMap + } + for (const doc of result.docs) { + const docRecord = doc as Record + for (const [key, value] of Object.entries(query)) { + if ( + (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') && + docRecord[key] == null + ) { + docRecord[key] = value + } + } + if (doc._class == null) { + doc._class = _class + } + } + return result + } + + async * iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + let cursor: string | undefined + do { + const page = await this.findAllPage(_class, query, { + ...options, + limit: options?.limit ?? 500, + cursor + }) + for (const doc of page.docs) { + yield doc + } + cursor = page.nextCursor + } while (cursor !== undefined) + } + private async checkRate (): Promise { if (this.currentRateLimit.remaining < this.currentRateLimit.limit / 3) { if (this.slowDownTimer < 50) { diff --git a/foundations/core/packages/api-client/src/rest/tx.ts b/foundations/core/packages/api-client/src/rest/tx.ts index d13a68ecea..592b16ade0 100644 --- a/foundations/core/packages/api-client/src/rest/tx.ts +++ b/foundations/core/packages/api-client/src/rest/tx.ts @@ -23,9 +23,12 @@ import { type DomainRequestOptions, type DomainResult, type FindOptions, + type FindPageOptions, + type FindPageResult, type FindResult, Hierarchy, ModelDb, + type IterateOptions, type OperationDomain, type Ref, type SearchOptions, @@ -77,6 +80,37 @@ class RestTxClient implements Client { return toFindResult(result, data.total) } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + const page = await this.client.findAllPage(_class, query, options) + return { + ...page, + docs: page.docs.map((doc) => this.hierarchy.updateLookupMixin(_class, doc, options)) + } + } + + async * iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + let cursor: string | undefined + do { + const page = await this.findAllPage(_class, query, { + ...options, + limit: options?.limit ?? 500, + cursor + }) + for (const doc of page.docs) { + yield doc + } + cursor = page.nextCursor + } while (cursor !== undefined) + } + async domainRequest( domain: OperationDomain, params: DomainParams, diff --git a/foundations/core/packages/api-client/src/rest/types.ts b/foundations/core/packages/api-client/src/rest/types.ts index 0d8567ca0b..ec921ca383 100644 --- a/foundations/core/packages/api-client/src/rest/types.ts +++ b/foundations/core/packages/api-client/src/rest/types.ts @@ -26,8 +26,11 @@ import { type DomainRequestOptions, type DomainResult, type FindOptions, + type FindPageOptions, + type FindPageResult, type FulltextStorage, type Hierarchy, + type IterateOptions, type Mixin, type MixinData, type MixinUpdate, @@ -53,6 +56,18 @@ export interface RestClient extends Storage, FulltextStorage { options?: FindOptions ) => Promise | undefined> + findAllPage: ( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ) => Promise> + + iterateAll: ( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ) => AsyncIterable> + getModel: () => Promise<{ hierarchy: Hierarchy, model: ModelDb }> domainRequest: ( diff --git a/foundations/core/packages/api-client/src/types.ts b/foundations/core/packages/api-client/src/types.ts index 0b509ceee1..086d7af464 100644 --- a/foundations/core/packages/api-client/src/types.ts +++ b/foundations/core/packages/api-client/src/types.ts @@ -25,12 +25,15 @@ import { type DocumentQuery, type DocumentUpdate, type FindOptions, + type FindPageOptions, + type FindPageResult, type FindResult, type Hierarchy, type Mixin, type MixinData, type MixinUpdate, type ModelDb, + type IterateOptions, type Ref, type Space, type TxResult, @@ -78,6 +81,18 @@ export interface FindOperations { options?: FindOptions | undefined ) => Promise> + findAllPage: ( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ) => Promise> + + iterateAll: ( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ) => AsyncIterable> + findOne: ( _class: Ref>, query: DocumentQuery, diff --git a/foundations/core/packages/client-resources/src/connection.ts b/foundations/core/packages/client-resources/src/connection.ts index 6ed03048e9..c79893a181 100644 --- a/foundations/core/packages/client-resources/src/connection.ts +++ b/foundations/core/packages/client-resources/src/connection.ts @@ -36,6 +36,8 @@ import core, { type DomainRequestOptions, type DomainResult, FindOptions, + FindPageOptions, + FindPageResult, FindResult, generateId, platformNow, @@ -956,6 +958,49 @@ class Connection implements ClientConnection { return result } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + const result = (await this.sendRequest({ + method: 'findAllPage', + params: [_class, query, options] + })) as FindPageResult + + if (result.lookupMap !== undefined) { + for (const doc of result.docs) { + if (doc.$lookup !== undefined) { + const lookup = doc.$lookup as Record + for (const [key, value] of Object.entries(lookup)) { + if (Array.isArray(value)) { + lookup[key] = value.map((item) => result.lookupMap?.[item]) + } else { + lookup[key] = result.lookupMap[value as string] + } + } + } + } + delete result.lookupMap + } + + for (const doc of result.docs) { + const docRecord = doc as Record + for (const [key, value] of Object.entries(query)) { + if ( + (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') && + docRecord[key] == null + ) { + docRecord[key] = value + } + } + if (doc._class == null) { + doc._class = _class + } + } + return result + } + tx (tx: Tx): Promise { return this.sendRequest({ method: 'tx', diff --git a/foundations/core/packages/core/src/__tests__/memdb.test.ts b/foundations/core/packages/core/src/__tests__/memdb.test.ts index ae17e9ef2c..c2896ac9b4 100644 --- a/foundations/core/packages/core/src/__tests__/memdb.test.ts +++ b/foundations/core/packages/core/src/__tests__/memdb.test.ts @@ -22,6 +22,9 @@ import { TxOperations } from '../operations' import { type DocumentQuery, type FindOptions, + type FindPageOptions, + type FindPageResult, + type IterateOptions, type SearchOptions, type SearchQuery, type SearchResult, @@ -52,6 +55,41 @@ class ClientModel extends ModelDb implements Client { return (await this.findAll(_class, query, options)).shift() } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + const { cursor, limit, ...findOptions } = options + const docs = await this.findAll(_class, query, findOptions) + const offset = cursor === undefined ? 0 : Number.parseInt(cursor, 10) + const start = Number.isNaN(offset) ? 0 : offset + const pageDocs = docs.slice(start, start + limit) + const nextOffset = start + pageDocs.length + return { + docs: pageDocs, + nextCursor: nextOffset < docs.length ? String(nextOffset) : undefined, + total: options.total === true ? docs.length : undefined + } + } + + async * iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + let cursor: string | undefined + do { + const page = await this.findAllPage(_class, query, { + ...options, + limit: options?.limit ?? 500, + cursor + }) + yield * page.docs + cursor = page.nextCursor + } while (cursor !== undefined) + } + async searchFulltext (query: SearchQuery, options: SearchOptions): Promise { return { docs: [] } } diff --git a/foundations/core/packages/core/src/client.ts b/foundations/core/packages/core/src/client.ts index 2226f60d6f..6f8cfef228 100644 --- a/foundations/core/packages/core/src/client.ts +++ b/foundations/core/packages/core/src/client.ts @@ -33,8 +33,11 @@ import type { DomainParams, DomainResult, FindOptions, + FindPageOptions, + FindPageResult, FindResult, FulltextStorage, + IterateOptions, SearchOptions, SearchQuery, SearchResult, @@ -65,6 +68,16 @@ export interface Client extends Storage, FulltextStorage { query: DocumentQuery, options?: FindOptions ) => Promise | undefined> + findAllPage: ( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ) => Promise> + iterateAll: ( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ) => AsyncIterable> close: () => Promise domainRequest: ( @@ -112,6 +125,11 @@ export interface ClientConnection extends Storage, FulltextStorage, BackupClient loadModel: (last: Timestamp, hash?: string) => Promise getLastHash?: (ctx: MeasureContext) => Promise pushHandler: (handler: TxHandler) => void + findAllPage?: ( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ) => Promise> domainRequest: (ctx: OperationDomain, params: DomainParams, options?: DomainRequestOptions) => Promise } @@ -159,6 +177,81 @@ class ClientImpl implements Client, BackupClient { return toFindResult(result, data.total) } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + const domain = this.hierarchy.getDomain(_class) + const data = + domain === DOMAIN_MODEL + ? await this.findModelPage(_class, query, options) + : this.conn.findAllPage !== undefined + ? await this.conn.findAllPage(_class, query, options) + : await this.findConnectionPage(_class, query, options) + + return { + ...data, + docs: data.docs.map((doc) => this.hierarchy.updateLookupMixin(_class, doc, options)) + } + } + + async * iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + let cursor: string | undefined + do { + const page = await this.findAllPage(_class, query, { + ...options, + limit: options?.limit ?? 500, + cursor + }) + for (const doc of page.docs) { + yield doc + } + cursor = page.nextCursor + } while (cursor !== undefined) + } + + private async findModelPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + const { cursor, limit, ...findOptions } = options + const sort: FindOptions['sort'] = options.sort ?? { _id: 1 } + const docs = await this.model.findAll(_class, query, { ...findOptions, sort }) + const offset = cursor === undefined ? 0 : Number.parseInt(cursor, 10) + const start = Number.isNaN(offset) || offset < 0 ? 0 : offset + const pageDocs = docs.slice(start, start + limit) + const nextOffset = start + pageDocs.length + return { + docs: pageDocs, + nextCursor: nextOffset < docs.length ? String(nextOffset) : undefined, + total: options.total === true ? docs.length : undefined + } + } + + private async findConnectionPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + const { cursor, limit, ...findOptions } = options + const docs = await this.conn.findAll(_class, query, findOptions) + const offset = cursor === undefined ? 0 : Number.parseInt(cursor, 10) + const start = Number.isNaN(offset) || offset < 0 ? 0 : offset + const pageDocs = docs.slice(start, start + limit) + const nextOffset = start + pageDocs.length + return { + docs: pageDocs, + nextCursor: nextOffset < docs.length ? String(nextOffset) : undefined, + total: options.total === true ? docs.length : undefined + } + } + async searchFulltext (query: SearchQuery, options: SearchOptions): Promise { return await this.conn.searchFulltext(query, options) } diff --git a/foundations/core/packages/core/src/operations.ts b/foundations/core/packages/core/src/operations.ts index 8e64f87928..697ffb5655 100644 --- a/foundations/core/packages/core/src/operations.ts +++ b/foundations/core/packages/core/src/operations.ts @@ -31,7 +31,10 @@ import type { DomainParams, DomainResult, FindOptions, + FindPageOptions, + FindPageResult, FindResult, + IterateOptions, SearchOptions, SearchQuery, SearchResult, @@ -78,6 +81,22 @@ export class TxOperations implements Omit { return this.client.findAll(_class, query, options) } + findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + return this.client.findAllPage(_class, query, options) + } + + iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + return this.client.iterateAll(_class, query, options) + } + findOne( _class: Ref>, query: DocumentQuery, @@ -474,6 +493,8 @@ export class ApplyOperations extends TxOperations { close: () => ops.client.close(), findOne: (_class, query, options?) => ops.client.findOne(_class, query, options), findAll: (_class, query, options?) => ops.client.findAll(_class, query, options), + findAllPage: (_class, query, options) => ops.client.findAllPage(_class, query, options), + iterateAll: (_class, query, options?) => ops.client.iterateAll(_class, query, options), searchFulltext: (query, options) => ops.client.searchFulltext(query, options), domainRequest: (domain, params) => ops.client.domainRequest(domain, params), tx: async (tx): Promise => { @@ -577,6 +598,8 @@ export class TxBuilder extends TxOperations { close: async () => {}, findOne: async (_class, query, options?) => undefined, findAll: async (_class, query, options?) => toFindResult([]), + findAllPage: async (_class, query, options) => ({ docs: [] }), + iterateAll: async function * () {}, searchFulltext: async (query, options) => ({ docs: [] }), domainRequest: async (domain, params) => ({ domain, value: null as any }), tx: async (tx): Promise => { diff --git a/foundations/core/packages/core/src/storage.ts b/foundations/core/packages/core/src/storage.ts index 06e1fb0f99..95d9aaaff1 100644 --- a/foundations/core/packages/core/src/storage.ts +++ b/foundations/core/packages/core/src/storage.ts @@ -221,6 +221,47 @@ export type FindResult = WithLookup[] & { lookupMap?: Record } +/** + * A deterministic scalar sort supported by cursor pagination. + * + * @public + */ +export type PaginationSortingQuery = { + [P in keyof T]?: SortingOrder +} & Record + +/** + * Options for keyset-based document pagination. + * + * @public + */ +export interface FindPageOptions extends Omit, 'limit' | 'sort'> { + limit: number + cursor?: string + sort?: PaginationSortingQuery +} + +/** + * A page returned by keyset-based document pagination. + * + * @public + */ +export interface FindPageResult { + docs: WithLookup[] + nextCursor?: string + total?: number + lookupMap?: Record +} + +/** + * Options for iterating over all matching documents page by page. + * + * @public + */ +export interface IterateOptions extends Omit, 'cursor' | 'limit'> { + limit?: number +} + export type DomainParams = Record export interface DomainResult { diff --git a/foundations/core/packages/query/src/__tests__/connection.ts b/foundations/core/packages/query/src/__tests__/connection.ts index 6c3acc7b65..d217ff4ea0 100644 --- a/foundations/core/packages/query/src/__tests__/connection.ts +++ b/foundations/core/packages/query/src/__tests__/connection.ts @@ -25,6 +25,8 @@ import core, { Domain, DOMAIN_TX, FindOptions, + FindPageOptions, + FindPageResult, FindResult, FulltextStorage, generateId, @@ -42,8 +44,10 @@ import core, { type DomainParams, type DomainRequestOptions, type DomainResult, + type IterateOptions, type OperationDomain, - type TxHandler + type TxHandler, + type WithLookup } from '@hcengineering/core' import { genMinModel } from './minmodel' @@ -103,6 +107,41 @@ FulltextStorage & { return (await this.findAll(_class, query, { ...options, limit: 1 })).shift() } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + const { cursor, limit, ...findOptions } = options + const result = await this.findAll(_class, query, findOptions) + const parsedOffset = cursor === undefined ? 0 : Number.parseInt(cursor, 10) + const offset = Number.isNaN(parsedOffset) || parsedOffset < 0 ? 0 : parsedOffset + const docs = result.slice(offset, offset + limit) + const nextOffset = offset + docs.length + return { + docs, + ...(nextOffset < result.length ? { nextCursor: String(nextOffset) } : {}), + ...(options.total === true ? { total: result.length } : {}) + } + } + + async * iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + let cursor: string | undefined + do { + const page = await this.findAllPage(_class, query, { + ...options, + limit: options?.limit ?? 500, + cursor + }) + yield * page.docs + cursor = page.nextCursor + } while (cursor !== undefined) + } + async domainRequest ( domain: OperationDomain, params: DomainParams, diff --git a/foundations/core/packages/query/src/index.ts b/foundations/core/packages/query/src/index.ts index b611830bfb..4ad4e1d9f4 100644 --- a/foundations/core/packages/query/src/index.ts +++ b/foundations/core/packages/query/src/index.ts @@ -24,6 +24,8 @@ import core, { Doc, DocumentQuery, FindOptions, + FindPageOptions, + FindPageResult, FindResult, Hierarchy, IndexingUpdateEvent, @@ -64,6 +66,7 @@ import core, { type DomainParams, type DomainRequestOptions, type DomainResult, + type IterateOptions, type OperationDomain } from '@hcengineering/core' import { PlatformError } from '@hcengineering/platform' @@ -293,6 +296,22 @@ export class LiveQuery implements WithTx, Client { return toFindResult(q.result.getClone(), q.total) } + findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + return this.client.findAllPage(_class, query, options) + } + + iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + return this.client.iterateAll(_class, query, options) + } + async domainRequest( domain: OperationDomain, params: DomainParams, diff --git a/foundations/server/packages/core/src/__tests__/pagination.test.ts b/foundations/server/packages/core/src/__tests__/pagination.test.ts new file mode 100644 index 0000000000..425c462154 --- /dev/null +++ b/foundations/server/packages/core/src/__tests__/pagination.test.ts @@ -0,0 +1,244 @@ +// +// Copyright © 2026 TraceX SAS. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import core, { toFindResult, type Doc, type FindResult, type Ref } from '@hcengineering/core' +import { PlatformError } from '@hcengineering/platform' +import { + buildCursorPayload, + calculatePageQueryHash, + checkCursorPayload, + filterByPagination, + findPage, + plainCursorCodec, + preparePageProjection, + preparePaginationFields +} from '../pagination' +import type { FindPagination } from '../types' + +interface TestDoc extends Doc { + name: string + flag?: boolean +} + +const testClass = core.class.Doc as Ref + +function doc (id: string, name: string, flag?: boolean): TestDoc { + return { + _id: id as Ref, + _class: testClass, + space: core.space.Space, + modifiedBy: core.account.System, + modifiedOn: 0, + name, + flag + } +} + +describe('pagination fields', () => { + it('appends _id as a tie breaker', () => { + expect(preparePaginationFields(testClass, { name: 1 })).toEqual([ + { field: 'name', order: 1 }, + { field: '_id', order: 1 } + ]) + }) + + it('keeps an explicit _id order', () => { + expect(preparePaginationFields(testClass, { _id: -1 })).toEqual([{ field: '_id', order: -1 }]) + }) + + it.each([{ '$lookup.space.name': 1 }, { 'nested.field': 1 }, { name: 5 as any }])( + 'rejects an unsupported sort %p', + (sort) => { + expect(() => preparePaginationFields(testClass, sort as any)).toThrow(PlatformError) + } + ) +}) + +describe('cursor payload', () => { + const fields = preparePaginationFields(testClass, { name: 1 }) + const hash = calculatePageQueryHash(testClass, { name: 'a' }, fields, undefined) + + it('survives a round trip', () => { + const payload = buildCursorPayload(testClass, hash, fields, doc('id-1', 'a')) + const decoded = plainCursorCodec.decode(plainCursorCodec.encode(payload)) + expect(checkCursorPayload(decoded, testClass, hash, fields)).toEqual(['a', 'id-1']) + }) + + it('is rejected when the query changes', () => { + const payload = buildCursorPayload(testClass, hash, fields, doc('id-1', 'a')) + const otherHash = calculatePageQueryHash(testClass, { name: 'b' }, fields, undefined) + expect(() => checkCursorPayload(payload, testClass, otherHash, fields)).toThrow(PlatformError) + }) + + it('is rejected when the sort changes', () => { + const payload = buildCursorPayload(testClass, hash, fields, doc('id-1', 'a')) + const otherFields = preparePaginationFields(testClass, { name: -1 }) + expect(() => checkCursorPayload(payload, testClass, hash, otherFields)).toThrow(PlatformError) + }) + + it('does not accept a foreign cursor', () => { + expect(() => plainCursorCodec.decode('not-a-cursor')).toThrow() + }) +}) + +describe('in memory keyset filter', () => { + const docs = [doc('a', 'one'), doc('b', 'two'), doc('c', 'three')] + + it('keeps documents after the cursor position', () => { + const pagination: FindPagination = { + fields: [{ field: '_id', order: 1 }], + values: ['a'] + } + expect(filterByPagination(docs, pagination).map(({ _id }) => _id)).toEqual(['b', 'c']) + }) + + it('follows a descending order', () => { + const pagination: FindPagination = { + fields: [{ field: '_id', order: -1 }], + values: ['c'] + } + expect(filterByPagination(docs, pagination).map(({ _id }) => _id)).toEqual(['a', 'b']) + }) + + it('uses the next field once the previous one is equal', () => { + const sameName = [doc('a', 'one'), doc('b', 'one'), doc('c', 'two')] + const pagination: FindPagination = { + fields: [ + { field: 'name', order: 1 }, + { field: '_id', order: 1 } + ], + values: ['one', 'a'] + } + expect(filterByPagination(sameName, pagination).map(({ _id }) => _id)).toEqual(['b', 'c']) + }) + + it('places nulls first for an ascending order', () => { + const withNulls = [doc('a', 'one'), doc('b', 'two', false), doc('c', 'three', true)] + const pagination: FindPagination = { + fields: [{ field: 'flag', order: 1 }], + values: [undefined] + } + expect(filterByPagination(withNulls, pagination).map(({ _id }) => _id)).toEqual(['b', 'c']) + }) + + it.each([1, -1] as const)('follows the in memory sort order, order %i', (order) => { + // resultSort compares strings with localeCompare, which differs from a code point order for mixed case + // values. A keyset predicate built on a different order would skip documents between pages. + const names = [ + 'Default Test Management', + 'Default Trainings', + 'Default teamspace type', + 'Default drive type', + 'Default product type', + 'Spaces' + ] + const sorted = names + .map((name, index) => doc(`id-${index}`, name)) + .sort((left, right) => left.name.localeCompare(right.name) * order) + + const collected: string[] = [] + let values: unknown[] | undefined + for (let page = 0; page <= names.length; page++) { + const rest = filterByPagination(sorted, { fields: [{ field: 'name', order }], values }) + if (rest.length === 0) { + break + } + collected.push(rest[0].name) + values = [rest[0].name] + } + expect(collected).toEqual(sorted.map(({ name }) => name)) + }) + + it('places nulls last for a descending order', () => { + const withNulls = [doc('a', 'one'), doc('b', 'two', false), doc('c', 'three', true)] + const pagination: FindPagination = { + fields: [{ field: 'flag', order: -1 }], + values: [true] + } + expect(filterByPagination(withNulls, pagination).map(({ _id }) => _id)).toEqual(['a', 'b']) + }) +}) + +describe('page projection', () => { + const fields = preparePaginationFields(testClass, { name: 1 }) + + it('adds cursor fields to an inclusion projection', () => { + const { projection, added } = preparePageProjection({ flag: 1 } as any, fields) + expect(projection).toEqual({ flag: 1, name: 1, _id: 1 }) + expect([...added]).toEqual(['name', '_id']) + }) + + it('removes cursor fields from an exclusion projection', () => { + const { projection, added } = preparePageProjection({ name: 0, flag: 0 } as any, fields) + expect(projection).toEqual({ flag: 0 }) + expect([...added]).toEqual(['name']) + }) +}) + +describe('findPage', () => { + const docs = [doc('a', 'one'), doc('b', 'two'), doc('c', 'three'), doc('d', 'four'), doc('e', 'five')] + + async function fetchPages (limit: number): Promise<{ pages: number, ids: string[] }> { + const ids: string[] = [] + let cursor: string | undefined + let pages = 0 + do { + const page = await findPage( + testClass, + {}, + { limit, cursor }, + async (pagination, sort, projection, pageLimit): Promise> => { + const sorted = [...docs].sort((left, right) => left._id.localeCompare(right._id)) + const filtered = filterByPagination(sorted, pagination) + return toFindResult(filtered.slice(0, pageLimit), filtered.length) + } + ) + ids.push(...page.docs.map(({ _id }) => _id)) + cursor = page.nextCursor + pages++ + } while (cursor !== undefined) + return { pages, ids } + } + + it('walks all pages without gaps or duplicates', async () => { + await expect(fetchPages(2)).resolves.toEqual({ pages: 3, ids: ['a', 'b', 'c', 'd', 'e'] }) + }) + + it('returns a single page when everything fits', async () => { + await expect(fetchPages(5)).resolves.toEqual({ pages: 1, ids: ['a', 'b', 'c', 'd', 'e'] }) + }) + + it.each([0, -1, 1.5, 1001])('rejects limit %p', async (limit) => { + await expect(fetchPages(limit)).rejects.toThrow(PlatformError) + }) + + it('rejects lookup based filters', async () => { + await expect( + findPage(testClass, { '$lookup.space.name': 'x' } as any, { limit: 10 }, async () => toFindResult([])) + ).rejects.toThrow(PlatformError) + }) + + it('rejects a cursor issued for another query', async () => { + const first = await findPage(testClass, {}, { limit: 2 }, async (pagination, sort, projection, limit) => + toFindResult(filterByPagination(docs, pagination).slice(0, limit), docs.length) + ) + expect(first.nextCursor).toBeDefined() + await expect( + findPage(testClass, { name: 'other' }, { limit: 2, cursor: first.nextCursor }, async () => + toFindResult([]) + ) + ).rejects.toThrow(PlatformError) + }) +}) diff --git a/foundations/server/packages/core/src/base.ts b/foundations/server/packages/core/src/base.ts index 0e929a2865..36d3faa507 100644 --- a/foundations/server/packages/core/src/base.ts +++ b/foundations/server/packages/core/src/base.ts @@ -21,7 +21,6 @@ import { type DocumentQuery, type Domain, type DomainParams, - type FindOptions, type FindResult, type LoadModelResponse, type MeasureContext, @@ -34,7 +33,7 @@ import { type Timestamp, type Tx } from '@hcengineering/core' -import type { Middleware, PipelineContext, TxMiddlewareResult } from './types' +import type { Middleware, PipelineContext, ServerFindOptions, TxMiddlewareResult } from './types' export const emptyFindResult = Promise.resolve(toFindResult([])) export const emptySearchResult = Promise.resolve({ docs: [], total: 0 }) @@ -55,7 +54,7 @@ export abstract class BaseMiddleware implements Middleware { ctx: MeasureContext, _class: Ref>, query: DocumentQuery, - options?: FindOptions + options?: ServerFindOptions ): Promise> { return this.provideFindAll(ctx, _class, query, options) } @@ -126,7 +125,7 @@ export abstract class BaseMiddleware implements Middleware { ctx: MeasureContext, _class: Ref>, query: DocumentQuery, - options?: FindOptions + options?: ServerFindOptions ): Promise> { if (this.next !== undefined) { return this.next.findAll(ctx, _class, query, options) diff --git a/foundations/server/packages/core/src/index.ts b/foundations/server/packages/core/src/index.ts index 8ca7cb2739..e79fee2d6b 100644 --- a/foundations/server/packages/core/src/index.ts +++ b/foundations/server/packages/core/src/index.ts @@ -21,6 +21,7 @@ export * from './benchmark' export * from './configuration' export * from './limitter' export * from './mem' +export * from './pagination' export * from './pipeline' export { default, serverCoreId } from './plugin' export * from './storage' diff --git a/foundations/server/packages/core/src/mem.ts b/foundations/server/packages/core/src/mem.ts index a8a7542979..511dd92009 100644 --- a/foundations/server/packages/core/src/mem.ts +++ b/foundations/server/packages/core/src/mem.ts @@ -36,6 +36,8 @@ import core, { type WorkspaceIds } from '@hcengineering/core' import { type DbAdapter, type DbAdapterHandler, type DomainHelperOperations, type RawFindIterator } from './adapter' +import { paginateInMemoryAsync } from './pagination' +import type { ServerFindOptions } from './types' /** * @public */ @@ -146,9 +148,18 @@ class InMemoryAdapter extends DummyDbAdapter implements DbAdapter { ctx: MeasureContext, _class: Ref>, query: DocumentQuery, - options?: FindOptions + options?: ServerFindOptions ): Promise> { - return ctx.withSync('inmem-find', {}, () => this.modeldb.findAll(_class, query, options)) + // A cursor position can not be pushed down to an in memory model, so it is applied in memory. + return ctx.with( + 'inmem-find', + {}, + async () => + await paginateInMemoryAsync( + options, + async (findOptions) => await this.modeldb.findAll(_class, query, findOptions) + ) + ) } load (ctx: MeasureContext, domain: Domain, docs: Ref[]): Promise { diff --git a/foundations/server/packages/core/src/pagination.ts b/foundations/server/packages/core/src/pagination.ts new file mode 100644 index 0000000000..d9073966b7 --- /dev/null +++ b/foundations/server/packages/core/src/pagination.ts @@ -0,0 +1,456 @@ +// +// Copyright © 2026 TraceX SAS. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import core, { + type Class, + type Doc, + type DocumentQuery, + type FindOptions, + type FindPageOptions, + type FindPageResult, + type FindResult, + type Hierarchy, + type Ref, + toFindResult +} from '@hcengineering/core' +import platform, { PlatformError, Severity, Status } from '@hcengineering/platform' +import { createHash } from 'crypto' +import type { FindPagination, FindPaginationField, ServerFindOptions } from './types' + +/** + * @public + */ +export const MAX_PAGE_SIZE = 1000 + +const CURSOR_VERSION = 1 + +/** + * A decoded cursor position. Kept implementation agnostic: `ClientSession` wraps it into a signed token, + * in-process clients encode it as plain JSON, but both validate it the very same way. + * + * @public + */ +export interface PageCursorPayload { + version: number + objectClass: string + queryHash: string + fields: FindPaginationField[] + values: unknown[] +} + +/** + * @public + */ +export function badPageRequest (): PlatformError> { + return new PlatformError(new Status(Severity.ERROR, platform.status.BadRequest, {})) +} + +/** + * @public + */ +export function stableStringify (value: unknown): string { + if (Array.isArray(value)) { + return `[${value.map((item) => stableStringify(item)).join(',')}]` + } + if (value !== null && typeof value === 'object') { + return `{${Object.entries(value) + .sort(([left], [right]) => left.localeCompare(right)) + .map(([key, item]) => `${JSON.stringify(key)}:${stableStringify(item)}`) + .join(',')}}` + } + return JSON.stringify(value) +} + +/** + * @public + */ +export function checkPageLimit (limit: number): void { + if (!Number.isInteger(limit) || limit < 1 || limit > MAX_PAGE_SIZE) { + throw badPageRequest() + } +} + +/** + * A cursor signature can not cover `options.lookup`: the very same `$lookup.*` filter key may be resolved + * through a different join on the next page (another target class, domain, or array/single lookup), so the + * signature stays valid while the result set changes, producing gaps or duplicates between pages. + * Lookup based filters are therefore not supported by the paged API, the same way as lookup based sorting. + * + * @public + */ +export function checkPageQuery (query: DocumentQuery): void { + for (const key of Object.keys(query)) { + if (key.startsWith('$lookup')) { + throw badPageRequest() + } + } +} + +/** + * Normalizes the requested sort into a deterministic list of cursor fields, appending `_id` as a tie breaker. + * When `hierarchy` is passed, attribute types not supported by keyset comparison are rejected as well. + * + * @public + */ +export function preparePaginationFields ( + _class: Ref>, + sort: FindPageOptions['sort'], + hierarchy?: Hierarchy +): FindPaginationField[] { + const fields: FindPaginationField[] = [] + for (const [field, value] of Object.entries(sort ?? {})) { + if (field.startsWith('$lookup') || (value !== 1 && value !== -1)) { + throw badPageRequest() + } + fields.push({ field, order: value }) + } + if (!fields.some(({ field }) => field === '_id')) { + fields.push({ field: '_id', order: 1 }) + } + + for (const { field } of fields) { + if (field.includes('.') || field.includes('$')) { + throw badPageRequest() + } + if (field === '_id' || hierarchy === undefined) { + continue + } + try { + const attr = hierarchy.findAttribute(_class, field) + if ( + attr === undefined || + attr.type._class === core.class.ArrOf || + attr.type._class === core.class.TypeIdentifier || + attr.type._class === core.class.EnumOf + ) { + throw badPageRequest() + } + } catch (err) { + if (err instanceof PlatformError) { + throw err + } + throw badPageRequest() + } + } + return fields +} + +/** + * @public + */ +export function calculatePageQueryHash ( + _class: Ref>, + query: DocumentQuery, + fields: FindPaginationField[], + showArchived: boolean | undefined +): string { + return createHash('sha256').update(stableStringify({ _class, query, fields, showArchived })).digest('base64url') +} + +/** + * Validates that a cursor was issued for the very same class, query and sort, and returns its position. + * + * @public + */ +export function checkCursorPayload ( + payload: PageCursorPayload, + _class: Ref>, + queryHash: string, + fields: FindPaginationField[] +): unknown[] { + if ( + payload.version !== CURSOR_VERSION || + payload.objectClass !== _class || + payload.queryHash !== queryHash || + stableStringify(payload.fields) !== stableStringify(fields) || + !Array.isArray(payload.values) || + payload.values.length !== fields.length + ) { + throw badPageRequest() + } + return payload.values +} + +/** + * @public + */ +export function buildCursorPayload ( + _class: Ref>, + queryHash: string, + fields: FindPaginationField[], + lastDoc: T +): PageCursorPayload { + const doc = lastDoc as unknown as Record + return { + version: CURSOR_VERSION, + objectClass: _class, + queryHash, + fields, + values: fields.map(({ field }) => doc[field]) + } +} + +/** + * @public + */ +export interface CursorCodec { + encode: (payload: PageCursorPayload) => string + decode: (cursor: string) => PageCursorPayload +} + +/** + * @public + */ +export interface FindPageContext { + // Enables validation of cursor field types. + hierarchy?: Hierarchy + // Defaults to a plain, unsigned codec suitable for in-process clients only. + codec?: CursorCodec +} + +/** + * A codec for in-process clients. Transactor sessions sign the very same payload with a token instead, + * so cursors are not interchangeable between the two and a foreign cursor is rejected rather than misread. + * + * @public + */ +export const plainCursorCodec: CursorCodec = { + encode: (payload) => Buffer.from(JSON.stringify(payload), 'utf8').toString('base64url'), + decode: (cursor) => JSON.parse(Buffer.from(cursor, 'base64url').toString('utf8')) as PageCursorPayload +} + +/** + * Cursor fields may be excluded by a projection, while their values are required to build the next cursor. + * Returns a patched projection along with the fields which have to be dropped from the response. + * + * @public + */ +export function preparePageProjection ( + projection: FindOptions['projection'], + fields: FindPaginationField[] +): { projection: FindOptions['projection'], added: Set } { + const added = new Set() + if (projection === undefined) { + return { projection, added } + } + const patched: Record = { ...(projection as Record) } + const inclusion = Object.values(patched).some((value) => value === 1) + for (const { field } of fields) { + if (inclusion && patched[field] !== 1) { + patched[field] = 1 + added.add(field) + } else if (!inclusion && patched[field] === 0) { + Reflect.deleteProperty(patched, field) + added.add(field) + } + } + return { projection: patched as FindOptions['projection'], added } +} + +/** + * @public + */ +export function dropAddedProjection (docs: T[], added: Set): void { + if (added.size === 0) { + return + } + for (const doc of docs as Array>) { + for (const field of added) { + Reflect.deleteProperty(doc, field) + } + } +} + +/** + * Shared keyset pagination flow for in-process clients: validates the request and the cursor, delegates a single + * fetch to the caller and builds the next cursor out of the last returned document. + * + * Transactor sessions follow the very same flow, but sign cursors with a token, so a client can not forge + * a position it has no access to. + * + * @public + */ +export async function findPage ( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions, + fetch: ( + pagination: FindPagination, + sort: FindOptions['sort'], + projection: FindOptions['projection'], + limit: number + ) => Promise>, + context: FindPageContext = {} +): Promise> { + const codec = context.codec ?? plainCursorCodec + checkPageLimit(options.limit) + checkPageQuery(query) + + const fields = preparePaginationFields(_class, options.sort, context.hierarchy) + const queryHash = calculatePageQueryHash(_class, query, fields, options.showArchived) + let values: unknown[] | undefined + if (options.cursor !== undefined) { + let payload: PageCursorPayload + try { + payload = codec.decode(options.cursor) + } catch (err) { + if (err instanceof PlatformError) { + throw err + } + throw badPageRequest() + } + values = checkCursorPayload(payload, _class, queryHash, fields) + } + + const { projection, added } = preparePageProjection(options.projection, fields) + const sort = Object.fromEntries(fields.map(({ field, order }) => [field, order])) as FindOptions['sort'] + + const result = await fetch({ fields, values }, sort, projection, options.limit + 1) + + const hasMore = result.length > options.limit + const docs = result.slice(0, options.limit) + const nextCursor = + hasMore && docs.length > 0 + ? codec.encode(buildCursorPayload(_class, queryHash, fields, docs[docs.length - 1])) + : undefined + dropAddedProjection(docs, added) + + return { + docs, + ...(nextCursor !== undefined ? { nextCursor } : {}), + ...(options.total === true ? { total: result.total } : {}), + ...(result.lookupMap !== undefined ? { lookupMap: result.lookupMap } : {}) + } +} + +/** + * Mirrors the comparison used by `resultSort` in core: in memory sorting compares strings with `localeCompare`, + * which is not a code point order, so the keyset predicate has to follow the very same rule. Otherwise a page + * boundary would be evaluated against an order the documents are not actually sorted by, skipping documents. + */ +function compareValues (left: unknown, right: unknown): number { + if (typeof left === 'string' && typeof right === 'string') { + return left.localeCompare(right) + } + if (left === right) { + return 0 + } + return (left as any) < (right as any) ? -1 : 1 +} + +/** + * Checks whether a document is positioned strictly after the cursor. + * + * Nulls follow the ordering used by database adapters: ascending sorts place them first, descending sorts last. + * + * @public + */ +export function isAfterCursor (doc: T, pagination: FindPagination): boolean { + const values = pagination.values + if (values === undefined) { + return true + } + const record = doc as unknown as Record + for (let index = 0; index < pagination.fields.length; index++) { + const { field, order } = pagination.fields[index] + const value = values[index] + const docValue = record[field] + const valueIsNull = value == null + const docIsNull = docValue == null + + if (valueIsNull && docIsNull) { + continue + } + if (order === 1) { + // Ascending: nulls come first. + if (valueIsNull) { + return true + } + if (docIsNull) { + return false + } + const compared = compareValues(docValue, value) + if (compared === 0) { + continue + } + return compared > 0 + } + // Descending: nulls come last. + if (valueIsNull) { + return false + } + if (docIsNull) { + return true + } + const compared = compareValues(docValue, value) + if (compared === 0) { + continue + } + return compared < 0 + } + // All cursor fields are equal, the document is the cursor itself. + return false +} + +/** + * In memory counterpart of the keyset predicate built by database adapters. Used by adapters which + * can not push the comparison down to a query. + * + * @public + */ +export function filterByPagination (docs: T[], pagination: FindPagination): T[] { + if (pagination.values === undefined) { + return docs + } + return docs.filter((doc) => isAfterCursor(doc, pagination)) +} + +function slicePage (all: FindResult, pagination: FindPagination, limit?: number): FindResult { + const filtered = filterByPagination(all, pagination) + return toFindResult(limit === undefined ? filtered : filtered.slice(0, limit), all.total, all.lookupMap) +} + +/** + * Applies a cursor position in memory, for storages which can not push the comparison down to a query. + * A limit is dropped from the passed options and applied to the filtered result instead. + * + * @public + */ +export function paginateInMemory ( + options: ServerFindOptions | undefined, + find: (findOptions?: ServerFindOptions) => FindResult +): FindResult { + const pagination = options?.pagination + if (pagination?.values === undefined) { + return find(options) + } + const { limit, ...findOptions } = options ?? {} + return slicePage(find(findOptions), pagination, limit) +} + +/** + * @public + */ +export async function paginateInMemoryAsync ( + options: ServerFindOptions | undefined, + find: (findOptions?: ServerFindOptions) => Promise> +): Promise> { + const pagination = options?.pagination + if (pagination?.values === undefined) { + return await find(options) + } + const { limit, ...findOptions } = options ?? {} + return slicePage(await find(findOptions), pagination, limit) +} diff --git a/foundations/server/packages/core/src/pipeline.ts b/foundations/server/packages/core/src/pipeline.ts index f71cbc101c..31763f75b1 100644 --- a/foundations/server/packages/core/src/pipeline.ts +++ b/foundations/server/packages/core/src/pipeline.ts @@ -23,7 +23,6 @@ import { type Domain, type DomainParams, type DomainResult, - type FindOptions, type FindResult, type LoadModelResponse, type MeasureContext, @@ -38,7 +37,13 @@ import { type TxResult } from '@hcengineering/core' import { emptyBroadcastResult } from './base' -import { type Middleware, type MiddlewareCreator, type Pipeline, type PipelineContext } from './types' +import { + type Middleware, + type MiddlewareCreator, + type Pipeline, + type PipelineContext, + type ServerFindOptions +} from './types' /** * @public @@ -101,7 +106,7 @@ class PipelineImpl implements Pipeline { ctx: MeasureContext, _class: Ref>, query: DocumentQuery, - options?: FindOptions + options?: ServerFindOptions ): Promise> { return this.head?.findAll(ctx, _class, query, options) ?? Promise.resolve(toFindResult([])) } diff --git a/foundations/server/packages/core/src/types.ts b/foundations/server/packages/core/src/types.ts index 09d7053c09..7d8adb1bb1 100644 --- a/foundations/server/packages/core/src/types.ts +++ b/foundations/server/packages/core/src/types.ts @@ -25,6 +25,8 @@ import { type DomainParams, type DomainResult, type FindOptions, + type FindPageOptions, + type FindPageResult, type FindResult, type Hierarchy, type LoadModelResponse, @@ -75,6 +77,19 @@ export interface ServerFindOptions extends FindOptions { memoryLimit?: number // in bytes // A bulk size for cursor fetching bulkSize?: number + + // A validated keyset position used by database adapters. + pagination?: FindPagination +} + +export interface FindPaginationField { + field: string + order: 1 | -1 +} + +export interface FindPagination { + fields: FindPaginationField[] + values?: unknown[] } export type SessionFindAll = ( @@ -217,7 +232,7 @@ export interface Pipeline { ctx: MeasureContext, _class: Ref>, query: DocumentQuery, - options?: FindOptions + options?: ServerFindOptions ) => Promise> searchFulltext: ( ctx: MeasureContext, @@ -633,6 +648,18 @@ export interface Session { query: DocumentQuery, options?: FindOptions ) => Promise> + findAllPage: ( + ctx: ClientSessionCtx, + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ) => Promise + findAllPageRaw: ( + ctx: ClientSessionCtx, + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ) => Promise> searchFulltext: (ctx: ClientSessionCtx, query: SearchQuery, options: SearchOptions) => Promise searchFulltextRaw: (ctx: ClientSessionCtx, query: SearchQuery, options: SearchOptions) => Promise tx: (ctx: ClientSessionCtx, tx: Tx) => Promise diff --git a/foundations/server/packages/core/src/utils.ts b/foundations/server/packages/core/src/utils.ts index 6f78ced48a..c9586d9149 100644 --- a/foundations/server/packages/core/src/utils.ts +++ b/foundations/server/packages/core/src/utils.ts @@ -22,6 +22,8 @@ import core, { type DomainParams, type DomainResult, type FindOptions, + type FindPageOptions, + type FindPageResult, type FindResult, type MeasureContext, type ModelDb, @@ -40,6 +42,7 @@ import { PlatformError, unknownError } from '@hcengineering/platform' import { createHash, type Hash } from 'crypto' import fs from 'fs' import type { DbAdapter } from './adapter' +import { findPage } from './pagination' import { BackupClientOps } from './storage' import type { OneSecondCounters, Pipeline } from './types' @@ -308,6 +311,39 @@ export function wrapPipeline ( result.total )[0] }, + findAllPage: async (_class, query, options) => { + const { cursor, limit, ...findOptions } = options + const page = await findPage( + _class, + query, + options, + async (pagination, sort, projection, pageLimit) => + await pipeline.findAll(ctx, _class, query, { + ...findOptions, + sort, + projection, + limit: pageLimit, + pagination + }), + { hierarchy: pipeline.context.hierarchy } + ) + return { + ...page, + docs: page.docs.map((doc) => pipeline.context.hierarchy.updateLookupMixin(_class, doc, options)) + } + }, + iterateAll: async function * (_class, query, options) { + let cursor: string | undefined + do { + const page = await this.findAllPage(_class, query, { + ...options, + limit: options?.limit ?? 500, + cursor + }) + yield * page.docs + cursor = page.nextCursor + } while (cursor !== undefined) + }, domainRequest: async (domain, params) => { return await pipeline.domainRequest(ctx, domain, params) }, @@ -360,6 +396,27 @@ export function wrapAdapterToClient (ctx: MeasureContext, storageAdapter: DbAdap return (await storageAdapter.findAll(ctx, _class, query, options)) as any } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + const { cursor, limit, ...findOptions } = options + return await findPage( + _class, + query, + options, + async (pagination, sort, projection, pageLimit) => + await storageAdapter.findAll(ctx, _class, query, { + ...findOptions, + sort, + projection, + limit: pageLimit, + pagination + }) + ) + } + async domainRequest(domain: OperationDomain, params: DomainParams): Promise> { return { domain, value: null as any } } diff --git a/foundations/server/packages/middleware/src/domainFind.ts b/foundations/server/packages/middleware/src/domainFind.ts index 2dcff85629..d5b1b996bf 100644 --- a/foundations/server/packages/middleware/src/domainFind.ts +++ b/foundations/server/packages/middleware/src/domainFind.ts @@ -27,7 +27,7 @@ import { } from '@hcengineering/core' import { PlatformError, unknownError } from '@hcengineering/platform' import type { DBAdapterManager, Middleware, PipelineContext, ServerFindOptions } from '@hcengineering/server-core' -import { BaseMiddleware, emptyFindResult } from '@hcengineering/server-core' +import { BaseMiddleware, emptyFindResult, paginateInMemory } from '@hcengineering/server-core' /** * Will perform a find inside adapters @@ -63,7 +63,10 @@ export class DomainFindMiddleware extends BaseMiddleware implements Middleware { const p = options?.prefix ?? 'client' const domain = this.context.hierarchy.getDomain(_class) if (domain === DOMAIN_MODEL) { - return Promise.resolve(this.context.modelDb.findAllSync(_class, query, options)) + // The model is served from memory, so a cursor position has to be applied here as well. + return Promise.resolve( + paginateInMemory(options, (findOptions) => this.context.modelDb.findAllSync(_class, query, findOptions)) + ) } return ctx.with( p + '-find-all', diff --git a/foundations/server/packages/middleware/src/findSecurity.ts b/foundations/server/packages/middleware/src/findSecurity.ts index b4f4732e00..e2d1e818bc 100644 --- a/foundations/server/packages/middleware/src/findSecurity.ts +++ b/foundations/server/packages/middleware/src/findSecurity.ts @@ -16,13 +16,17 @@ import { type Class, type Doc, type DocumentQuery, - type FindOptions, type FindResult, type MeasureContext, type Ref, type SessionData } from '@hcengineering/core' -import { BaseMiddleware, type Middleware, type PipelineContext } from '@hcengineering/server-core' +import { + BaseMiddleware, + type Middleware, + type PipelineContext, + type ServerFindOptions +} from '@hcengineering/server-core' /** * @public @@ -44,10 +48,10 @@ export class FindSecurityMiddleware extends BaseMiddleware implements Middleware ctx: MeasureContext, _class: Ref>, query: DocumentQuery, - options?: FindOptions + options?: ServerFindOptions ): Promise> { if (options != null) { - const { limit, sort, lookup, projection, associations, total, showArchived } = options + const { limit, sort, lookup, projection, associations, total, showArchived, pagination } = options return this.provideFindAll(ctx, _class, query, { limit, sort, @@ -55,7 +59,8 @@ export class FindSecurityMiddleware extends BaseMiddleware implements Middleware projection, associations, total, - showArchived + showArchived, + pagination }) } return this.provideFindAll(ctx, _class, query, options) diff --git a/foundations/server/packages/middleware/src/liveQuery.ts b/foundations/server/packages/middleware/src/liveQuery.ts index f76d8dfda2..794ff127e6 100644 --- a/foundations/server/packages/middleware/src/liveQuery.ts +++ b/foundations/server/packages/middleware/src/liveQuery.ts @@ -61,6 +61,23 @@ export class LiveQueryMiddleware extends BaseMiddleware implements Middleware { results.total ) }, + findAllPage: async (_class, query, options) => { + const { cursor, limit, ...findOptions } = options + const results = await this.findAll(metrics, _class, query, findOptions) + const offset = cursor === undefined ? 0 : Number.parseInt(cursor, 10) + const start = Number.isNaN(offset) ? 0 : offset + const docs = results.slice(start, start + limit) + const nextOffset = start + docs.length + return { + docs, + nextCursor: nextOffset < results.length ? String(nextOffset) : undefined, + total: options.total === true ? results.length : undefined + } + }, + iterateAll: async function * (_class, query, options) { + const results = await this.findAll(_class, query, options) + yield * results + }, findOne: async (_class, query, options) => { const _ctx: MeasureContext = (options as ServerFindOptions)?.ctx ?? metrics delete (options as ServerFindOptions)?.ctx diff --git a/foundations/server/packages/middleware/src/lookup.ts b/foundations/server/packages/middleware/src/lookup.ts index 7b5f1da4ad..ebf3aaf096 100644 --- a/foundations/server/packages/middleware/src/lookup.ts +++ b/foundations/server/packages/middleware/src/lookup.ts @@ -17,14 +17,18 @@ import { type Class, type Doc, type DocumentQuery, - type FindOptions, type FindResult, type MeasureContext, type Ref, clone, toFindResult } from '@hcengineering/core' -import { BaseMiddleware, type Middleware, type PipelineContext } from '@hcengineering/server-core' +import { + BaseMiddleware, + type Middleware, + type PipelineContext, + type ServerFindOptions +} from '@hcengineering/server-core' /** * @public */ @@ -45,7 +49,7 @@ export class LookupMiddleware extends BaseMiddleware implements Middleware { ctx: MeasureContext, _class: Ref>, query: DocumentQuery, - options?: FindOptions + options?: ServerFindOptions ): Promise> { const result = await this.provideFindAll(ctx, _class, query, options) // Fill lookup map to make more compact representation @@ -83,23 +87,32 @@ export class LookupMiddleware extends BaseMiddleware implements Middleware { } } const lookupMap = Object.fromEntries(Array.from(Object.values(idClassMap)).map((it) => [it.id, it.doc])) - return this.cleanQuery(toFindResult(newResult, result.total, lookupMap), query, lookupMap) + return this.cleanQuery( + toFindResult(newResult, result.total, lookupMap), + query, + lookupMap, + new Set(options.pagination?.fields.map(({ field }) => field)) + ) } // We need to get rid of simple query parameters matched in documents - return this.cleanQuery(result, query) + return this.cleanQuery(result, query, undefined, new Set(options?.pagination?.fields.map(({ field }) => field))) } private cleanQuery( result: FindResult, query: DocumentQuery, - lookupMap?: Record + lookupMap?: Record, + preserveKeys = new Set() ): FindResult { const newResult: T[] = [] for (const doc of result) { let _doc = doc let cloned = false for (const [k, v] of Object.entries(query)) { + if (preserveKeys.has(k)) { + continue + } if (typeof v === 'string' || typeof v === 'number' || typeof v === 'boolean') { if ((_doc as any)[k] === v) { if (!cloned) { diff --git a/foundations/server/packages/middleware/src/model.ts b/foundations/server/packages/middleware/src/model.ts index f6bd24ced0..2388afd385 100644 --- a/foundations/server/packages/middleware/src/model.ts +++ b/foundations/server/packages/middleware/src/model.ts @@ -17,7 +17,6 @@ import core, { type Class, type Doc, type DocumentQuery, - type FindOptions, type FindResult, type Hierarchy, type LoadModelResponse, @@ -36,10 +35,11 @@ import type { Middleware, MiddlewareCreator, PipelineContext, + ServerFindOptions, TxAdapter, TxMiddlewareResult } from '@hcengineering/server-core' -import { BaseMiddleware } from '@hcengineering/server-core' +import { BaseMiddleware, paginateInMemoryAsync } from '@hcengineering/server-core' import crypto from 'node:crypto' const isAccountTx = (it: TxCUD): boolean => @@ -90,11 +90,15 @@ export class ModelMiddleware extends BaseMiddleware implements Middleware { ctx: MeasureContext, _class: Ref>, query: DocumentQuery, - options?: FindOptions + options?: ServerFindOptions ): Promise> { const d = this.context.hierarchy.findDomain(_class) if (d === DOMAIN_MODEL) { - return this.context.modelDb.findAll(_class, query, options) + // The model is served from memory, so a cursor position has to be applied here as well. + return paginateInMemoryAsync( + options, + async (findOptions) => await this.context.modelDb.findAll(_class, query, findOptions) + ) } return this.provideFindAll(ctx, _class, query, options) } diff --git a/foundations/server/packages/mongo/src/storage.ts b/foundations/server/packages/mongo/src/storage.ts index 4c8b8513d1..0e219e073e 100644 --- a/foundations/server/packages/mongo/src/storage.ts +++ b/foundations/server/packages/mongo/src/storage.ts @@ -688,6 +688,10 @@ abstract class MongoAdapterBase implements DbAdapter { } } const totalPipeline: any[] = [...pipeline] + const paginationQuery = this.buildPaginationQuery(clazz, options) + if (paginationQuery !== undefined) { + pipeline.push({ $match: paginationQuery }) + } this.fillSortPipeline(clazz, options, pipeline) if (options?.limit !== undefined || typeof query._id === 'string') { pipeline.push({ $limit: options?.limit ?? 1 }) @@ -909,6 +913,8 @@ abstract class MongoAdapterBase implements DbAdapter { const stTime = platformNow() const mongoQuery = this.translateQuery(_class, query, options) const fQuery = { ...mongoQuery.base, ...mongoQuery.lookup } + const paginationQuery = this.buildPaginationQuery(_class, options) + const pageQuery = paginationQuery === undefined ? fQuery : { $and: [fQuery, paginationQuery] } return addOperation(ctx, 'find-all', {}, async () => { const st = platformNow() let result: FindResult @@ -946,7 +952,7 @@ abstract class MongoAdapterBase implements DbAdapter { doc = null } } else { - doc = await coll.findOne(fQuery, findOptions) + doc = await coll.findOne(pageQuery, findOptions) } let total = -1 @@ -962,7 +968,7 @@ abstract class MongoAdapterBase implements DbAdapter { ) } - let cursor = coll.find(fQuery) + let cursor = coll.find(pageQuery) if (options?.projection !== undefined) { const projection = this.calcProjection(options, _class) @@ -1047,6 +1053,40 @@ abstract class MongoAdapterBase implements DbAdapter { return sort } + private buildPaginationQuery( + _class: Ref>, + options?: ServerFindOptions + ): Filter | undefined { + const pagination = options?.pagination + if (pagination?.values === undefined) { + return + } + const branches: Array> = [] + for (let index = 0; index < pagination.fields.length; index++) { + const branch: Record = {} + for (let prefix = 0; prefix < index; prefix++) { + const field = pagination.fields[prefix] + branch[this.translateKey(field.field, _class).key] = pagination.values[prefix] + } + const field = pagination.fields[index] + const key = this.translateKey(field.field, _class).key + const value = pagination.values[index] + if (value == null) { + if (field.order === 1) { + branch[key] = { $ne: null } + branches.push(branch) + } + continue + } + branch[key] = { [field.order === 1 ? '$gt' : '$lt']: value } + branches.push(branch) + if (field.order === -1) { + branches.push({ ...branch, [key]: null }) + } + } + return { $or: branches } + } + private calcProjection( options: | (FindOptions & { diff --git a/foundations/server/packages/postgres/src/storage.ts b/foundations/server/packages/postgres/src/storage.ts index c2fec29622..2850c4bcc4 100644 --- a/foundations/server/packages/postgres/src/storage.ts +++ b/foundations/server/packages/postgres/src/storage.ts @@ -477,7 +477,9 @@ abstract class PostgresAdapterBase implements DbAdapter { if (joins.length > 0) { sqlChunks.push(this.buildJoinString(vars, joins)) } - sqlChunks.push(`WHERE ${this.buildQuery(vars, _class, domain, query, joins, options)}`) + const baseQuery = this.buildQuery(vars, _class, domain, query, joins, options) + const paginationQuery = this.buildPaginationQuery(vars, _class, domain, joins, options) + sqlChunks.push(`WHERE ${baseQuery}${paginationQuery === undefined ? '' : ` AND (${paginationQuery})`}`) const showArchived = shouldShowArchived(query, options) const secJoin = this.addSecurity(_class, vars, query, showArchived, domain, ctx.contextData) @@ -485,7 +487,7 @@ abstract class PostgresAdapterBase implements DbAdapter { sqlChunks.push(secJoin) } if (options?.sort !== undefined) { - sqlChunks.push(this.buildOrder(_class, domain, options.sort, joins)) + sqlChunks.push(this.buildOrder(_class, domain, options.sort, joins, options.pagination !== undefined)) } if (options?.limit !== undefined) { sqlChunks.push(`LIMIT ${escape(options.limit)}`) @@ -1043,7 +1045,8 @@ abstract class PostgresAdapterBase implements DbAdapter { _class: Ref>, baseDomain: string, sort: SortingQuery, - joins: JoinProps[] + joins: JoinProps[], + pagination: boolean = false ): string { const res: string[] = [] for (const _key in sort) { @@ -1055,7 +1058,11 @@ abstract class PostgresAdapterBase implements DbAdapter { if (typeof val === 'number') { const key = escape(_key) if (attr !== undefined && NumericTypes.includes(attr.type._class)) { - res.push(`(${this.getKey(_class, baseDomain, key, joins)})::numeric ${val === 1 ? 'ASC' : 'DESC'}`) + res.push( + `(${this.getKey(_class, baseDomain, key, joins)})::numeric ${val === 1 ? 'ASC' : 'DESC'}${ + pagination ? (val === 1 ? ' NULLS FIRST' : ' NULLS LAST') : '' + }` + ) } else if (attr !== undefined && attr.type._class === core.class.TypeIdentifier) { res.push( `regexp_replace(COALESCE(${this.getKey(_class, baseDomain, key, joins)}, ''), '-?\\d+$', '') ${val === 1 ? 'ASC' : 'DESC'}` @@ -1073,7 +1080,11 @@ abstract class PostgresAdapterBase implements DbAdapter { ) } } else { - res.push(`${this.getKey(_class, baseDomain, key, joins)} ${val === 1 ? 'ASC' : 'DESC'}`) + res.push( + `${this.getKey(_class, baseDomain, key, joins)} ${val === 1 ? 'ASC' : 'DESC'}${ + pagination ? (val === 1 ? ' NULLS FIRST' : ' NULLS LAST') : '' + }` + ) } } else { // todo handle custom sorting @@ -1086,6 +1097,88 @@ abstract class PostgresAdapterBase implements DbAdapter { } } + private buildPaginationQuery( + vars: ValuesVariables, + _class: Ref>, + baseDomain: string, + joins: JoinProps[], + options?: ServerFindOptions + ): string | undefined { + const pagination = options?.pagination + if (pagination?.values === undefined) { + return + } + + const branches: string[] = [] + for (let index = 0; index < pagination.fields.length; index++) { + const equalities: string[] = [] + for (let prefix = 0; prefix < index; prefix++) { + const field = pagination.fields[prefix] + const { expression, value } = this.getPaginationTerm( + _class, + baseDomain, + field.field, + joins, + pagination.values[prefix] + ) + equalities.push(value == null ? `${expression} IS NULL` : `${expression} = ${vars.add(value)}`) + } + + const field = pagination.fields[index] + const { expression, value } = this.getPaginationTerm( + _class, + baseDomain, + field.field, + joins, + pagination.values[index] + ) + let comparison: string + if (value == null) { + comparison = field.order === 1 ? `${expression} IS NOT NULL` : 'FALSE' + } else if (field.order === 1) { + comparison = `${expression} > ${vars.add(value)}` + } else { + comparison = `(${expression} < ${vars.add(value)} OR ${expression} IS NULL)` + } + branches.push([...equalities, comparison].join(' AND ')) + } + return branches.join(' OR ') + } + + /** + * Builds a keyset comparison term: an SQL expression for a cursor field along with a value + * normalized to the type the expression yields. + */ + private getPaginationTerm( + _class: Ref>, + baseDomain: string, + field: string, + joins: JoinProps[], + value: unknown + ): { expression: string, value: unknown } { + const attr = this.hierarchy.findAttribute(_class, field) + if ( + attr?.type._class === core.class.TypeIdentifier || + attr?.type._class === core.class.EnumOf || + attr?.type._class === core.class.ArrOf + ) { + throw new Error(`Unsupported cursor sort field: ${field}`) + } + const expression = this.getKey(_class, baseDomain, escape(field), joins) + if (attr !== undefined && NumericTypes.includes(attr.type._class)) { + // Both sides are numeric, buildOrder casts the very same expression as well. + return { expression: `(${expression})::numeric`, value } + } + // Attributes kept in JSONB are extracted as `text`, so a cursor value has to be compared as text too, + // the same way regular query values are normalized in translateQueryValue. Casting the expression instead + // would desync the keyset predicate from ORDER BY, which sorts these fields as text. + const isJsonText = expression.includes('data') && (expression.includes('->') || expression.includes('#>>')) + if (isJsonText && (typeof value === 'boolean' || typeof value === 'number' || typeof value === 'bigint')) { + return { expression, value: `${value}` } + } + return { expression, value } + } + private buildQuery( vars: ValuesVariables, _class: Ref>, diff --git a/foundations/server/packages/server/src/client.ts b/foundations/server/packages/server/src/client.ts index b7770d41e1..3f3700d5b2 100644 --- a/foundations/server/packages/server/src/client.ts +++ b/foundations/server/packages/server/src/client.ts @@ -26,6 +26,8 @@ import { type DomainParams, type DomainResult, type FindOptions, + type FindPageOptions, + type FindPageResult, type FindResult, type LoadModelResponse, type MeasureContext, @@ -49,18 +51,22 @@ import { import { PlatformError, unknownError } from '@hcengineering/platform' import { BackupClientOps, + badPageRequest, createBroadcastEvent, estimateDocSize, + findPage, SessionDataImpl, type ClientSessionCtx, type ConnectionSocket, + type CursorCodec, type OneSecondCounters, + type PageCursorPayload, type Pipeline, type Session, type SessionRequest, type StatisticsElement } from '@hcengineering/server-core' -import { type Token } from '@hcengineering/server-token' +import { decodeToken, generateToken, type Token } from '@hcengineering/server-token' const useReserveContext = (process.env.USE_RESERVE_CTX ?? 'true') === 'true' @@ -232,6 +238,84 @@ export class ClientSession implements Session { } } + async findAllPageRaw( + ctx: ClientSessionCtx, + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + this.lastRequest = Date.now() + this.total.find++ + this.current.find++ + this.includeSessionContext(ctx) + + const findOptions = { ...options } + delete findOptions.cursor + return await findPage( + _class, + query, + options, + async (pagination, sort, projection, limit) => + await ctx.pipeline.findAll(ctx.ctx, _class, query, { + ...findOptions, + sort, + projection, + limit, + pagination + }), + { hierarchy: ctx.pipeline.context.hierarchy, codec: this.cursorCodec() } + ) + } + + /** + * Cursors handed out to a client are signed, so a client can not forge a position, reuse a cursor issued + * for another account or workspace, or pass a cursor produced by an in-process client. + */ + private cursorCodec (): CursorCodec { + return { + encode: (payload) => generateToken(this.account.uuid, this.workspace.uuid, { cursor: JSON.stringify(payload) }), + decode: (cursor) => { + const decoded = decodeToken(cursor) + if (decoded.account !== this.account.uuid || decoded.workspace !== this.workspace.uuid) { + throw badPageRequest() + } + return JSON.parse(decoded.extra?.cursor ?? '') as PageCursorPayload + } + } + } + + async findAllPage( + ctx: ClientSessionCtx, + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise { + const domain = ctx.pipeline.context.hierarchy.findDomain(_class) ?? '' + if (domain === '') { + await ctx.sendError( + ctx.requestId, + 'Invalid class name is passed. Failed to findAllPage.', + new Error('Unknown domain') + ) + return + } + try { + const result = await this.counter.withCounter('find-page-' + domain, 1, () => + this.findAllPageRaw(ctx, _class, query, options) + ) + await this.counter.withCounter('clientSendMemory', this.estimateSize(result), () => + ctx.sendResponse(ctx.requestId, result) + ) + } catch (err) { + await ctx.sendError( + ctx.requestId, + 'Failed to findAllPage', + err instanceof PlatformError ? err.status : unknownError(err) + ) + ctx.ctx.error('failed to findAllPage', { err }) + } + } + async searchFulltext (ctx: ClientSessionCtx, query: SearchQuery, options: SearchOptions): Promise { try { this.lastRequest = Date.now() diff --git a/packages/presentation/src/pipeline.ts b/packages/presentation/src/pipeline.ts index 6a6df593ab..20f1e341df 100644 --- a/packages/presentation/src/pipeline.ts +++ b/packages/presentation/src/pipeline.ts @@ -8,8 +8,11 @@ import { type DomainRequestOptions, type DomainResult, type FindOptions, + type FindPageOptions, + type FindPageResult, type FindResult, type Hierarchy, + type IterateOptions, type ModelDb, type OperationDomain, type QuerySelector, @@ -154,6 +157,22 @@ export class PresentationPipelineImpl implements PresentationPipeline { } } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + return await this.client.findAllPage(_class, query, options) + } + + iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + return this.client.iterateAll(_class, query, options) + } + async searchFulltext (query: SearchQuery, options: SearchOptions): Promise { return await this.client.searchFulltext(query, options) } diff --git a/packages/presentation/src/utils.ts b/packages/presentation/src/utils.ts index 6aa5303845..395ac538c7 100644 --- a/packages/presentation/src/utils.ts +++ b/packages/presentation/src/utils.ts @@ -30,11 +30,14 @@ import core, { type DomainRequestOptions, type DomainResult, type FindOptions, + type FindPageOptions, + type FindPageResult, type FindResult, getCurrentAccount, platformNow, hasAccountRole, type Hierarchy, + type IterateOptions, MeasureMetricsContext, type Mixin, type ModelDb, @@ -108,7 +111,7 @@ export const pendingCreatedDocs = writable, boolean>>({}) class UIClient extends TxOperations implements Client { constructor ( client: Client, - private readonly liveQuery: Client + private readonly liveQuery: Pick ) { super(client, getCurrentAccount().primarySocialId) } @@ -332,6 +335,22 @@ class ClientHookImpl implements Client { return await this.client.findAll(_class, query, options) } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + return await this.client.findAllPage(_class, query, options) + } + + iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + return this.client.iterateAll(_class, query, options) + } + async domainRequest( domain: OperationDomain, params: DomainParams, diff --git a/pods/media/src/client.ts b/pods/media/src/client.ts index f606c2d0b3..ccd9df02f8 100644 --- a/pods/media/src/client.ts +++ b/pods/media/src/client.ts @@ -25,8 +25,11 @@ import core, { DomainRequestOptions, DomainResult, FindOptions, + FindPageOptions, + FindPageResult, FindResult, Hierarchy, + IterateOptions, MeasureContext, ModelDb, OperationDomain, @@ -179,6 +182,22 @@ class RestClientAdapter implements Client { return await this.client.findAll(_class, query, options) } + async findAllPage( + _class: Ref>, + query: DocumentQuery, + options: FindPageOptions + ): Promise> { + return await this.client.findAllPage(_class, query, options) + } + + iterateAll( + _class: Ref>, + query: DocumentQuery, + options?: IterateOptions + ): AsyncIterable> { + return this.client.iterateAll(_class, query, options) + } + async tx (tx: Tx): Promise { return await this.client.tx(tx) } diff --git a/pods/server/src/rpc.ts b/pods/server/src/rpc.ts index 641adf57d3..be1bccde38 100644 --- a/pods/server/src/rpc.ts +++ b/pods/server/src/rpc.ts @@ -295,6 +295,26 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) }) + app.post('/api/v1/find-page/:workspaceId', (req, res) => { + void withSession(req, res, 'findAllPage', async (ctx, session, rateLimit) => { + const { _class, query, options }: any = (await retrieveJson(req)) ?? {} + + try { + const result = await session.findAllPageRaw(ctx, _class, query, options) + await sendJson(req, res, result, rateLimitToHeaders(rateLimit)) + } catch (err: unknown) { + if (err instanceof PlatformError && err.status.code === platform.status.BadRequest) { + sendError(res, 400, { + message: 'Failed to execute operation', + error: 'Invalid pagination request' + }) + return + } + throw err + } + }) + }) + app.post('/api/v1/tx/:workspaceId', (req, res) => { void withSession(req, res, 'tx', async (ctx, session, rateLimit) => { const tx: any = (await retrieveJson(req)) ?? {} diff --git a/ws-tests/api-tests/src/__tests__/cursor.test.ts b/ws-tests/api-tests/src/__tests__/cursor.test.ts new file mode 100644 index 0000000000..c7dab7444a --- /dev/null +++ b/ws-tests/api-tests/src/__tests__/cursor.test.ts @@ -0,0 +1,388 @@ +/** + Copyright © 2026 TraceX SAS. + + Licensed under the Eclipse Public License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. You may + obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + + See the License for the specific language governing permissions and + limitations under the License. +*/ + +import { + createRestClient, + createRestTxOperations, + getWorkspaceToken, + loadServerConfig, + type RestClient, + type ServerConfig, + type WorkspaceToken +} from '@hcengineering/api-client' +import { getClient as getAccountClient, type AccountClient } from '@hcengineering/account-client' +import chunter, { type Channel } from '@hcengineering/chunter' +import core, { + AccountRole, + generateId, + SortingOrder, + systemAccountUuid, + type Ref, + type TxOperations, + type WithLookup +} from '@hcengineering/core' +import { generateToken } from '@hcengineering/server-token' + +interface CollectedPages { + docs: Array> + pageCount: number + totals: number[] +} + +describe('cursor-api', () => { + const frontUrl = process.env.FRONT_URL ?? 'http://huly.local:8083' + const workspaceName = 'api-tests' + const runId = generateId() + const prefix = `cursor-${runId}` + const largeCount = 127 + + let config: ServerConfig + let ownerWorkspace: WorkspaceToken + let readerWorkspace: WorkspaceToken + let ownerClient: RestClient + let readerClient: RestClient + let fixtureClient: RestClient + let txOperations: TxOperations + const createdChannels: Array> = [] + const publicChannels: Array> = [] + const privateChannels: Array> = [] + + beforeAll(async () => { + config = await loadServerConfig(frontUrl) + ownerWorkspace = await getWorkspaceToken( + frontUrl, + { + email: 'user1', + password: '1234', + workspace: workspaceName + }, + config + ) + + try { + readerWorkspace = await getWorkspaceToken( + frontUrl, + { + email: 'user2', + password: '1234', + workspace: workspaceName + }, + config + ) + } catch { + const adminClient: AccountClient = getAccountClient( + config.ACCOUNTS_URL, + generateToken(systemAccountUuid, undefined, { service: 'workspace', admin: 'true' }, 'secret') + ) + await adminClient.assignWorkspace('user2', ownerWorkspace.workspaceId, AccountRole.User) + readerWorkspace = await getWorkspaceToken( + frontUrl, + { + email: 'user2', + password: '1234', + workspace: workspaceName + }, + config + ) + } + + ownerClient = createRestClient(ownerWorkspace.endpoint, ownerWorkspace.workspaceId, ownerWorkspace.token) + readerClient = createRestClient(readerWorkspace.endpoint, readerWorkspace.workspaceId, readerWorkspace.token) + fixtureClient = createRestClient( + ownerWorkspace.endpoint, + ownerWorkspace.workspaceId, + generateToken(systemAccountUuid, ownerWorkspace.workspaceId, undefined, 'secret') + ) + txOperations = await createRestTxOperations( + ownerWorkspace.endpoint, + ownerWorkspace.workspaceId, + ownerWorkspace.token + ) + + for (let start = 0; start < largeCount; start += 20) { + const end = Math.min(start + 20, largeCount) + await Promise.all( + Array.from({ length: end - start }, async (_, offset) => { + const index = start + offset + const isPrivate = index % 4 === 0 + // autoJoin is not a column of the space domain, it is kept in JSONB and extracted as text, + // so it covers cursors over a custom boolean field. + const autoJoin = index % 3 === 0 + const id = await fixtureClient.createDoc(chunter.class.Channel, core.space.Space, { + name: `${prefix}-${index.toString().padStart(3, '0')}`, + description: '', + private: isPrivate, + archived: false, + members: isPrivate ? [ownerWorkspace.info.account] : [], + autoJoin + }) + createdChannels.push(id) + if (isPrivate) { + privateChannels.push(id) + } else { + publicChannels.push(id) + } + }) + ) + } + }, 120000) + + afterAll(async () => { + if (fixtureClient === undefined) { + return + } + for (let start = 0; start < createdChannels.length; start += 20) { + const ids = createdChannels.slice(start, start + 20) + const docs = await fixtureClient.findAll(chunter.class.Channel, { _id: { $in: ids } }) + await Promise.all(docs.map(async (doc) => await fixtureClient.remove(doc))) + } + }, 120000) + + it('reads a large result without gaps or duplicates', async () => { + const { docs, pageCount, totals } = await collectPages(ownerClient, prefix, 17, true) + const ids = docs.map(({ _id }) => _id) + + expect(pageCount).toBe(8) + expect(totals).toEqual(Array.from({ length: pageCount }, () => largeCount)) + expect(ids).toHaveLength(largeCount) + expect(new Set(ids).size).toBe(largeCount) + expect(new Set(ids)).toEqual(new Set(createdChannels)) + expect(docs.map(({ name }) => name)).toEqual([...docs.map(({ name }) => name)].sort()) + }) + + it('uses _id as a stable tie breaker for duplicate sort values', async () => { + const docs: Array> = [] + let cursor: string | undefined + do { + const page = await ownerClient.findAllPage( + chunter.class.Channel, + { name: { $like: `${prefix}%` } }, + { + limit: 11, + cursor, + sort: { topic: SortingOrder.Ascending } + } + ) + docs.push(...page.docs) + cursor = page.nextCursor + } while (cursor !== undefined) + + expect(docs).toHaveLength(largeCount) + expect(new Set(docs.map(({ _id }) => _id))).toEqual(new Set(createdChannels)) + }) + + it.each([ + { count: 0, limit: 10, pages: 1 }, + { count: 1, limit: 10, pages: 1 }, + { count: 3, limit: 10, pages: 1 }, + { count: 10, limit: 10, pages: 1 }, + { count: 11, limit: 10, pages: 2 } + ])('handles a small result with $count documents', async ({ count, limit, pages }) => { + const ids = createdChannels.slice(0, count) + const result = await collectPagesByIds(ownerClient, ids, limit) + + expect(result.pageCount).toBe(pages) + expect(result.docs.map(({ _id }) => _id)).toHaveLength(count) + expect(new Set(result.docs.map(({ _id }) => _id))).toEqual(new Set(ids)) + }) + + it('does not expose private spaces while traversing pages', async () => { + const ownerResult = await collectPages(ownerClient, prefix, 9, true) + const readerResult = await collectPages(readerClient, prefix, 9, true) + const readerIds = new Set(readerResult.docs.map(({ _id }) => _id)) + + expect(new Set(ownerResult.docs.map(({ _id }) => _id))).toEqual(new Set(createdChannels)) + expect(readerIds).toEqual(new Set(publicChannels)) + for (const privateId of privateChannels) { + expect(readerIds.has(privateId)).toBe(false) + } + expect(readerResult.totals).toEqual(Array.from({ length: readerResult.pageCount }, () => publicChannels.length)) + }) + + it('rejects a cursor issued for another account', async () => { + const firstPage = await ownerClient.findAllPage( + chunter.class.Channel, + { name: { $like: `${prefix}%` } }, + { limit: 7, sort: { name: SortingOrder.Ascending } } + ) + + expect(firstPage.nextCursor).toBeDefined() + await expect( + readerClient.findAllPage( + chunter.class.Channel, + { name: { $like: `${prefix}%` } }, + { + limit: 7, + sort: { name: SortingOrder.Ascending }, + cursor: firstPage.nextCursor + } + ) + ).rejects.toThrow() + }) + + it('rejects a cursor reused with another query', async () => { + const firstPage = await ownerClient.findAllPage( + chunter.class.Channel, + { name: { $like: `${prefix}%` } }, + { limit: 7, sort: { name: SortingOrder.Ascending } } + ) + + expect(firstPage.nextCursor).toBeDefined() + await expect( + ownerClient.findAllPage( + chunter.class.Channel, + { _id: { $in: createdChannels.slice(0, 20) } }, + { + limit: 7, + sort: { name: SortingOrder.Ascending }, + cursor: firstPage.nextCursor + } + ) + ).rejects.toThrow() + }) + + it.each([SortingOrder.Ascending, SortingOrder.Descending])( + 'paginates over a boolean field kept in JSONB, order %i', + async (order) => { + const docs: Array> = [] + let cursor: string | undefined + do { + const page = await ownerClient.findAllPage( + chunter.class.Channel, + { name: { $like: `${prefix}%` } }, + { + limit: 10, + cursor, + sort: { autoJoin: order } + } + ) + docs.push(...page.docs) + cursor = page.nextCursor + } while (cursor !== undefined) + + expect(docs).toHaveLength(largeCount) + expect(new Set(docs.map(({ _id }) => _id))).toEqual(new Set(createdChannels)) + + const flags = docs.map(({ autoJoin }) => autoJoin === true) + const expectedFlags = [...flags].sort((left, right) => + left === right ? 0 : (left ? 1 : -1) * (order === SortingOrder.Ascending ? 1 : -1) + ) + expect(flags).toEqual(expectedFlags) + } + ) + + it('paginates over a model domain class', async () => { + // The model is served from memory and bypasses database adapters, so the cursor position has to be + // applied by the pipeline itself, otherwise every page repeats the very first one. + const ids: string[] = [] + let cursor: string | undefined + let pages = 0 + do { + const page = await ownerClient.findAllPage( + core.class.SpaceType, + {}, + { limit: 1, cursor, sort: { name: SortingOrder.Ascending } } + ) + ids.push(...page.docs.map(({ _id }) => _id)) + cursor = page.nextCursor + pages++ + expect(pages).toBeLessThan(100) + } while (cursor !== undefined) + + const all = await ownerClient.findAll(core.class.SpaceType, {}) + expect(new Set(ids)).toEqual(new Set(all.map(({ _id }) => _id))) + expect(ids).toHaveLength(all.length) + }) + + it('rejects lookup based filters', async () => { + await expect( + ownerClient.findAllPage(chunter.class.Channel, { '$lookup.space._id': core.space.Space } as any, { + limit: 7, + sort: { name: SortingOrder.Ascending }, + lookup: { space: core.class.Space } + }) + ).rejects.toThrow() + }) + + it('iterates over all pages', async () => { + const docs: Array> = [] + for await (const doc of txOperations.iterateAll( + chunter.class.Channel, + { name: { $like: `${prefix}%` } }, + { limit: 13, sort: { name: SortingOrder.Ascending } } + )) { + docs.push(doc) + } + + expect(new Set(docs.map(({ _id }) => _id))).toEqual(new Set(createdChannels)) + }) +}) + +async function collectPages ( + client: RestClient, + prefix: string, + limit: number, + total: boolean +): Promise { + const docs: Array> = [] + const totals: number[] = [] + let cursor: string | undefined + let pageCount = 0 + do { + const page = await client.findAllPage( + chunter.class.Channel, + { name: { $like: `${prefix}%` } }, + { + limit, + cursor, + total, + sort: { name: SortingOrder.Ascending } + } + ) + expect(page.docs.length).toBeLessThanOrEqual(limit) + if (page.nextCursor !== undefined) { + expect(page.docs).toHaveLength(limit) + } + if (total) { + expect(page.total).toBeGreaterThanOrEqual(page.docs.length) + totals.push(page.total ?? -1) + } + docs.push(...page.docs) + cursor = page.nextCursor + pageCount++ + } while (cursor !== undefined) + return { docs, pageCount, totals } +} + +async function collectPagesByIds (client: RestClient, ids: Array>, limit: number): Promise { + const docs: Array> = [] + let cursor: string | undefined + let pageCount = 0 + do { + const page = await client.findAllPage( + chunter.class.Channel, + { _id: { $in: ids } }, + { + limit, + cursor, + sort: { name: SortingOrder.Ascending } + } + ) + docs.push(...page.docs) + cursor = page.nextCursor + pageCount++ + } while (cursor !== undefined) + return { docs, pageCount, totals: [] } +}