Fix cursor pagination with optional tie-breaker (#2080)

ravikiranvm · web-flow · commit 173cc01354e9 · 2026-03-17T11:36:59.000+05:30
Fixes OPS-3810. ## Additional Notes We create benchmark workflows in bulk, so many rows share the same updated timestamp. With cursor pagination, page boundaries previously filtered using only updated < cursor, which caused rows with the same timestamp to be skipped across pages. This PR adds support for an optional secondary pagination key (used as a secondary sort key) to break ties when primary timestamps are identical.
diff --git a/packages/server/api/src/app/flows/flow/flow.service.ts b/packages/server/api/src/app/flows/flow/flow.service.ts
@@ -145,6 +145,11 @@ export const flowService = {
         columnName: 'fv.updated',
         columnType: 'timestamp with time zone',
       },
+      customPaginationSecondaryColumn: {
+        columnPath: 'id',
+        columnName: 'flow.id',
+        columnType: 'string',
+      },
     });
 
     const queryWhere: Record<string, unknown> = {
diff --git a/packages/server/api/src/app/helper/pagination/build-paginator.ts b/packages/server/api/src/app/helper/pagination/build-paginator.ts
@@ -8,16 +8,28 @@ export type PagingQuery = {
   order?: Order | 'ASC' | 'DESC';
 };
 
+type CustomPaginationColumnOptions = {
+  columnPath: string;
+  columnName: string;
+  columnType?: string;
+};
+
+// Secondary custom pagination is only valid when primary custom pagination is configured.
+type CustomPaginationColumns =
+  | {
+      customPaginationColumn?: never;
+      customPaginationSecondaryColumn?: never;
+    }
+  | {
+      customPaginationColumn: CustomPaginationColumnOptions;
+      customPaginationSecondaryColumn?: CustomPaginationColumnOptions;
+    };
+
 export type PaginationOptions<Entity> = {
   entity: EntitySchema<Entity>;
   alias?: string;
   query?: PagingQuery;
-  customPaginationColumn?: {
-    columnPath: string;
-    columnName: string;
-    columnType?: string;
-  };
-};
+} & CustomPaginationColumns;
 
 export function buildPaginator<Entity extends ObjectLiteral>(
   options: PaginationOptions<Entity>,
@@ -32,6 +44,15 @@ export function buildPaginator<Entity extends ObjectLiteral>(
 
   paginator.setAlias(alias);
 
+  if (
+    options.customPaginationSecondaryColumn &&
+    !options.customPaginationColumn
+  ) {
+    throw new Error(
+      'customPaginationSecondaryColumn requires customPaginationColumn',
+    );
+  }
+
   if (options.customPaginationColumn) {
     paginator.setPaginationColumn(
       options.customPaginationColumn.columnPath,
@@ -40,6 +61,14 @@ export function buildPaginator<Entity extends ObjectLiteral>(
     );
   }
 
+  if (options.customPaginationSecondaryColumn) {
+    paginator.setPaginationSecondaryColumn(
+      options.customPaginationSecondaryColumn.columnPath,
+      options.customPaginationSecondaryColumn.columnName,
+      options.customPaginationSecondaryColumn.columnType,
+    );
+  }
+
   if (query.afterCursor) {
     paginator.setAfterCursor(query.afterCursor);
   }
diff --git a/packages/server/api/src/app/helper/pagination/paginator.ts b/packages/server/api/src/app/helper/pagination/paginator.ts
@@ -31,8 +31,16 @@ export type PagingResult<Entity> = {
   cursor: CursorResult;
 };
 
+type CursorContext = {
+  primaryColumnName: string;
+  primaryParamName: string;
+  secondaryColumnName: string | null;
+  secondaryParamName: string | null;
+};
+
 const PAGINATION_KEY = 'created';
 const CUSTOM_PAGINATION_KEY = 'custom_pagination';
+const CUSTOM_PAGINATION_SECONDARY_KEY = 'custom_pagination_tie_breaker';
 const DEFAULT_TIMESTAMP_TYPE = 'timestamp with time zone';
 
 export default class Paginator<Entity extends ObjectLiteral> {
@@ -56,6 +64,12 @@ export default class Paginator<Entity extends ObjectLiteral> {
 
   private paginationColumnType: string | null = null;
 
+  private paginationSecondaryColumnPath: string | null = null;
+
+  private paginationSecondaryColumnName: string | null = null;
+
+  private paginationSecondaryColumnType: string | null = null;
+
   public constructor(private readonly entity: EntitySchema) {}
 
   public setPaginationColumn(
@@ -72,6 +86,16 @@ export default class Paginator<Entity extends ObjectLiteral> {
     this.alias = alias;
   }
 
+  public setPaginationSecondaryColumn(
+    columnPath: string,
+    columnName: string,
+    columnType = 'string',
+  ): void {
+    this.paginationSecondaryColumnPath = columnPath;
+    this.paginationSecondaryColumnName = columnName;
+    this.paginationSecondaryColumnType = columnType;
+  }
+
   public setAfterCursor(cursor: string): void {
     this.afterCursor = cursor;
   }
@@ -171,28 +195,26 @@ export default class Paginator<Entity extends ObjectLiteral> {
   ): void {
     const dbType = system.get(AppSystemProp.DB_TYPE);
     const operator = this.getOperator();
-    let queryString: string;
-
-    const isCustomColumn =
-      this.paginationColumnName && cursors[CUSTOM_PAGINATION_KEY];
-    const columnName = isCustomColumn
-      ? this.paginationColumnName
-      : `${this.alias}.${PAGINATION_KEY}`;
-    const paramName = isCustomColumn ? CUSTOM_PAGINATION_KEY : PAGINATION_KEY;
-
-    if (dbType === DatabaseType.SQLITE3) {
-      queryString = `${columnName} ${operator} :${paramName}`;
-    } else if (dbType === DatabaseType.POSTGRES) {
-      if (this.hasBeforeCursor() && !this.hasAfterCursor()) {
-        queryString = `${columnName} ${operator} (:${paramName}::timestamp + INTERVAL '1 millisecond')`;
-      } else {
-        queryString = `${columnName} ${operator} :${paramName}::timestamp`;
-      }
-    } else {
-      throw new Error('Unsupported database type');
+    const context = this.resolveCursorContext(cursors);
+
+    if (context.secondaryColumnName && context.secondaryParamName) {
+      this.applyCompositeCursorFilter(
+        where,
+        cursors,
+        dbType,
+        operator,
+        context,
+      );
+      return;
     }
 
-    where.orWhere(queryString, cursors);
+    this.applySingleColumnCursorFilter(
+      where,
+      cursors,
+      dbType,
+      operator,
+      context,
+    );
   }
 
   private getOperator(): string {
@@ -222,6 +244,10 @@ export default class Paginator<Entity extends ObjectLiteral> {
       orderByCondition[`${this.alias}.${PAGINATION_KEY}`] = order;
     }
 
+    if (this.paginationColumnName && this.paginationSecondaryColumnName) {
+      orderByCondition[this.paginationSecondaryColumnName] = order;
+    }
+
     return orderByCondition;
   }
 
@@ -257,9 +283,31 @@ export default class Paginator<Entity extends ObjectLiteral> {
       this.paginationColumnType || DEFAULT_TIMESTAMP_TYPE,
       value,
     );
-    const payload = `${CUSTOM_PAGINATION_KEY}:${encodedValue}`;
+    const payload = [`${CUSTOM_PAGINATION_KEY}:${encodedValue}`];
 
-    return btoa(payload);
+    if (
+      this.paginationSecondaryColumnPath &&
+      this.paginationSecondaryColumnName
+    ) {
+      const secondaryValue = getValueByPath(
+        entity,
+        this.paginationSecondaryColumnPath,
+      );
+      if (secondaryValue === null || secondaryValue === undefined) {
+        throw new Error(
+          `Pagination secondary column not found at path: ${this.paginationSecondaryColumnPath}`,
+        );
+      }
+      const encodedSecondaryValue = encodeByType(
+        this.paginationSecondaryColumnType || 'string',
+        secondaryValue,
+      );
+      payload.push(
+        `${CUSTOM_PAGINATION_SECONDARY_KEY}:${encodedSecondaryValue}`,
+      );
+    }
+
+    return btoa(payload.join(','));
   }
 
   private decode(cursor: string): CursorParam {
@@ -279,6 +327,9 @@ export default class Paginator<Entity extends ObjectLiteral> {
     if (key === CUSTOM_PAGINATION_KEY) {
       return this.paginationColumnType || DEFAULT_TIMESTAMP_TYPE;
     }
+    if (key === CUSTOM_PAGINATION_SECONDARY_KEY) {
+      return this.paginationSecondaryColumnType || 'string';
+    }
 
     const col = this.entity.options.columns[key];
     if (col === undefined) {
@@ -291,6 +342,154 @@ export default class Paginator<Entity extends ObjectLiteral> {
     return order === Order.ASC ? Order.DESC : Order.ASC;
   }
 
+  private buildComparisonClause({
+    dbType,
+    columnName,
+    paramName,
+    operator,
+  }: {
+    dbType: string | undefined;
+    columnName: string;
+    paramName: string;
+    operator: string;
+  }): string {
+    if (dbType === DatabaseType.SQLITE3) {
+      return `${columnName} ${operator} :${paramName}`;
+    }
+
+    if (dbType === DatabaseType.POSTGRES) {
+      const type = this.getEntityPropertyType(paramName);
+      if (this.isTimestampType(type)) {
+        if (operator === '<') {
+          return `${columnName} < :${paramName}::timestamptz`;
+        }
+        if (operator === '>') {
+          return `${columnName} >= (:${paramName}::timestamptz + INTERVAL '1 millisecond')`;
+        }
+        if (operator === '=') {
+          return `(${columnName} >= :${paramName}::timestamptz AND ${columnName} < (:${paramName}::timestamptz + INTERVAL '1 millisecond'))`;
+        }
+        return `${columnName} ${operator} :${paramName}::timestamptz`;
+      }
+      return `${columnName} ${operator} :${paramName}`;
+    }
+
+    throw new Error('Unsupported database type');
+  }
+
+  private isTimestampType(type: string): boolean {
+    return (
+      type === 'timestamp with time zone' ||
+      type === 'datetime' ||
+      type === 'date'
+    );
+  }
+
+  private resolveCursorContext(cursors: CursorParam): CursorContext {
+    const customPaginationColumnName = this.paginationColumnName;
+    const hasCustomPaginationCursor =
+      customPaginationColumnName !== null &&
+      cursors[CUSTOM_PAGINATION_KEY] !== undefined;
+
+    const primaryColumnName =
+      hasCustomPaginationCursor && customPaginationColumnName
+        ? customPaginationColumnName
+        : `${this.alias}.${PAGINATION_KEY}`;
+    const primaryParamName = hasCustomPaginationCursor
+      ? CUSTOM_PAGINATION_KEY
+      : PAGINATION_KEY;
+
+    const hasCustomSecondaryCursor =
+      this.paginationSecondaryColumnName !== null &&
+      cursors[CUSTOM_PAGINATION_SECONDARY_KEY] !== undefined;
+
+    if (hasCustomPaginationCursor && hasCustomSecondaryCursor) {
+      return {
+        primaryColumnName,
+        primaryParamName,
+        secondaryColumnName: this.paginationSecondaryColumnName,
+        secondaryParamName: CUSTOM_PAGINATION_SECONDARY_KEY,
+      };
+    }
+
+    return {
+      primaryColumnName,
+      primaryParamName,
+      secondaryColumnName: null,
+      secondaryParamName: null,
+    };
+  }
+
+  private applySingleColumnCursorFilter(
+    where: WhereExpressionBuilder,
+    cursors: CursorParam,
+    dbType: string | undefined,
+    operator: string,
+    context: CursorContext,
+  ): void {
+    where.orWhere(
+      this.buildComparisonClause({
+        dbType,
+        columnName: context.primaryColumnName,
+        paramName: context.primaryParamName,
+        operator,
+      }),
+      cursors,
+    );
+  }
+
+  private applyCompositeCursorFilter(
+    where: WhereExpressionBuilder,
+    cursors: CursorParam,
+    dbType: string | undefined,
+    operator: string,
+    context: CursorContext,
+  ): void {
+    const {
+      primaryColumnName,
+      primaryParamName,
+      secondaryColumnName,
+      secondaryParamName,
+    } = context;
+    if (!secondaryColumnName || !secondaryParamName) {
+      throw new Error('Pagination secondary context is not configured');
+    }
+
+    where.orWhere(
+      this.buildComparisonClause({
+        dbType,
+        columnName: primaryColumnName,
+        paramName: primaryParamName,
+        operator,
+      }),
+      cursors,
+    );
+
+    // Lexicographic cursor compare: primary equals, then compare secondary key.
+    where.orWhere(
+      new Brackets((nestedWhere) => {
+        nestedWhere.where(
+          this.buildComparisonClause({
+            dbType,
+            columnName: primaryColumnName,
+            paramName: primaryParamName,
+            operator: '=',
+          }),
+          cursors,
+        );
+        nestedWhere.andWhere(
+          this.buildComparisonClause({
+            dbType,
+            columnName: secondaryColumnName,
+            paramName: secondaryParamName,
+            operator,
+          }),
+          cursors,
+        );
+      }),
+    );
+  }
+
   private toPagingResult<Entity>(entities: Entity[]): PagingResult<Entity> {
     return {
       data: entities,
diff --git a/packages/server/api/test/integration/helper/pagination/paginator.integration.test.ts b/packages/server/api/test/integration/helper/pagination/paginator.integration.test.ts
