feat: add cursor-based pagination to all list endpoints (#1085)

Add opt-in cursor-based pagination to all list API endpoints:
messages, posts (v0/v1), aggregates, files, credit history,
balances, credit balances, and address stats (v1).

Cursors are opaque base64url-encoded JSON with ISO 8601 timestamps.
Pass cursor= (empty) to start cursor mode from the first page.
Page/offset mode is fully preserved for backward compatibility.

Author: odesenfans

src/aleph/db/accessors/aggregates.py

@@ -320,6 +320,10 @@ def get_aggregates(
     sort_order: SortOrder = SortOrder.DESCENDING,
     page: int = 1,
     pagination: int = 100,
+    after_time: Optional[dt.datetime] = None,
+    after_key: Optional[str] = None,
+    after_owner: Optional[str] = None,
+    cursor_mode: bool = False,
 ) -> Iterable[AggregateDb]:
     where_clause = []
     if keys:
@@ -328,15 +332,15 @@ def get_aggregates(
         where_clause.append(AggregateDb.owner.in_(addresses))
 
     if sort_by == SortByAggregate.CREATION_TIME:
-        order_by_column: Any = AggregateDb.creation_datetime
+        order_by_raw_column: Any = AggregateDb.creation_datetime
     else:
         # last_modified
-        order_by_column = AggregateElementDb.creation_datetime
+        order_by_raw_column = AggregateElementDb.creation_datetime
 
     if sort_order == SortOrder.DESCENDING:
-        order_by_column = order_by_column.desc()
+        order_by_column = order_by_raw_column.desc()
     else:
-        order_by_column = order_by_column.asc()
+        order_by_column = order_by_raw_column.asc()
 
     query = (
         select(AggregateDb)
@@ -346,10 +350,45 @@ def get_aggregates(
         )
         .where(*where_clause)
         .order_by(order_by_column)
-        .limit(pagination)
-        .offset((page - 1) * pagination)
     )
 
+    if after_time is not None:
+        if sort_order == SortOrder.DESCENDING:
+            query = query.where(
+                (order_by_raw_column < after_time)
+                | (
+                    (order_by_raw_column == after_time)
+                    & (
+                        (AggregateDb.key > after_key)
+                        | (
+                            (AggregateDb.key == after_key)
+                            & (AggregateDb.owner > after_owner)
+                        )
+                    )
+                )
+            )
+        else:
+            query = query.where(
+                (order_by_raw_column > after_time)
+                | (
+                    (order_by_raw_column == after_time)
+                    & (
+                        (AggregateDb.key > after_key)
+                        | (
+                            (AggregateDb.key == after_key)
+                            & (AggregateDb.owner > after_owner)
+                        )
+                    )
+                )
+            )
+    elif page > 1:
+        query = query.offset((page - 1) * pagination)
+
+    if pagination:
+        query = query.limit(
+            pagination + 1 if after_time is not None or cursor_mode else pagination
+        )
+
     return (
         session.execute(query.options(selectinload(AggregateDb.last_revision)))
     ).scalars()

src/aleph/db/accessors/balances.py

@@ -53,6 +53,8 @@ def make_balances_by_chain_query(
     page: int = 1,
     pagination: int = 100,
     min_balance: int = 0,
+    after_address: Optional[str] = None,
+    cursor_mode: bool = False,
 ) -> Select:
     query = select(AlephBalanceDb.address, AlephBalanceDb.balance, AlephBalanceDb.chain)
 
@@ -62,11 +64,19 @@ def make_balances_by_chain_query(
     if min_balance > 0:
         query = query.filter(AlephBalanceDb.balance >= min_balance)
 
-    query = query.offset((page - 1) * pagination)
+    query = query.order_by(AlephBalanceDb.address.asc())
 
-    # If pagination == 0, return all matching results
-    if pagination:
-        query = query.limit(pagination)
+    if after_address is not None:
+        query = query.where(AlephBalanceDb.address > after_address)
+
+    if after_address is not None or cursor_mode:
+        if pagination:
+            query = query.limit(pagination + 1)
+    else:
+        query = query.offset((page - 1) * pagination)
+        # If pagination == 0, return all matching results
+        if pagination:
+            query = query.limit(pagination)
 
     return query
 
@@ -387,6 +397,8 @@ def get_credit_balances(
     page: int = 1,
     pagination: int = 100,
     min_balance: int = 0,
+    after_address: Optional[str] = None,
+    cursor_mode: bool = False,
 ) -> list[tuple[str, int]]:
     """
     Get paginated credit balances for all addresses.
@@ -397,10 +409,18 @@ def get_credit_balances(
     if min_balance > 0:
         query = query.filter(AlephCreditBalanceDb.balance >= min_balance)
 
-    query = query.offset((page - 1) * pagination)
+    query = query.order_by(AlephCreditBalanceDb.address.asc())
+
+    if after_address is not None:
+        query = query.where(AlephCreditBalanceDb.address > after_address)
 
-    if pagination:
-        query = query.limit(pagination)
+    if after_address is not None or cursor_mode:
+        if pagination:
+            query = query.limit(pagination + 1)
+    else:
+        query = query.offset((page - 1) * pagination)
+        if pagination:
+            query = query.limit(pagination)
 
     # Return results in the expected format (address, credits)
     results = session.execute(query).all()
@@ -734,6 +754,10 @@ def get_address_credit_history(
     origin: Optional[str] = None,
     origin_ref: Optional[str] = None,
     payment_method: Optional[str] = None,
+    after_time: Optional[dt.datetime] = None,
+    after_credit_ref: Optional[str] = None,
+    after_credit_index: Optional[int] = None,
+    cursor_mode: bool = False,
 ) -> Sequence[AlephCreditHistoryDb]:
     """
     Get paginated credit history entries for a specific address, ordered from newest to oldest.
@@ -750,14 +774,21 @@ def get_address_credit_history(
         origin: Filter by origin
         origin_ref: Filter by origin reference
         payment_method: Filter by payment method
+        after_time: Cursor-based: only return entries older than this timestamp
+        after_credit_ref: Cursor-based: tiebreaker credit_ref for entries with the same timestamp
+        after_credit_index: Cursor-based: tiebreaker credit_index for entries with the same timestamp and credit_ref
 
     Returns:
         List of credit history entries ordered by message_timestamp desc
     """
     query = (
         select(AlephCreditHistoryDb)
         .where(AlephCreditHistoryDb.address == address)
-        .order_by(AlephCreditHistoryDb.message_timestamp.desc())
+        .order_by(
+            AlephCreditHistoryDb.message_timestamp.desc(),
+            AlephCreditHistoryDb.credit_ref.desc(),
+            AlephCreditHistoryDb.credit_index.desc(),
+        )
     )
 
     # Apply filters
@@ -776,7 +807,25 @@ def get_address_credit_history(
     if payment_method is not None:
         query = query.where(AlephCreditHistoryDb.payment_method == payment_method)
 
-    if pagination > 0:
+    if after_time is not None:
+        query = query.where(
+            (AlephCreditHistoryDb.message_timestamp < after_time)
+            | (
+                (AlephCreditHistoryDb.message_timestamp == after_time)
+                & (
+                    (AlephCreditHistoryDb.credit_ref < after_credit_ref)
+                    | (
+                        (AlephCreditHistoryDb.credit_ref == after_credit_ref)
+                        & (AlephCreditHistoryDb.credit_index < after_credit_index)
+                    )
+                )
+            )
+        )
+
+    if after_time is not None or cursor_mode:
+        if pagination > 0:
+            query = query.limit(pagination + 1)
+    elif pagination > 0:
         query = query.offset((page - 1) * pagination).limit(pagination)
 
     return session.execute(query).scalars().all()

src/aleph/db/accessors/files.py

@@ -234,6 +234,9 @@ def get_address_files_for_api(
     pagination: int = 0,
     page: int = 1,
     sort_order: SortOrder = SortOrder.DESCENDING,
+    after_time: Optional[dt.datetime] = None,
+    after_hash: Optional[str] = None,
+    cursor_mode: bool = False,
 ) -> Iterable[Row]:
     select_stmt = (
         select(
@@ -247,22 +250,47 @@ def get_address_files_for_api(
         .where(MessageFilePinDb.owner == owner)
     )
 
-    if pagination:
-        select_stmt = select_stmt.limit(pagination).offset((page - 1) * pagination)
-
     if sort_order == SortOrder.DESCENDING:
         order_by_columns: Tuple[UnaryExpression[Any], UnaryExpression[Any]] = (
             MessageFilePinDb.created.desc(),
             MessageFilePinDb.item_hash.asc(),
         )
     else:  # ASCENDING
         order_by_columns = (
-            MessageFilePinDb.item_hash.asc(),
+            MessageFilePinDb.created.asc(),
             MessageFilePinDb.item_hash.asc(),
         )
 
     select_stmt = select_stmt.order_by(*order_by_columns)
 
+    if after_time is not None:
+        if sort_order == SortOrder.DESCENDING:
+            select_stmt = select_stmt.where(
+                (MessageFilePinDb.created < after_time)
+                | (
+                    (MessageFilePinDb.created == after_time)
+                    & (MessageFilePinDb.item_hash > after_hash)
+                )
+            )
+        else:
+            select_stmt = select_stmt.where(
+                (MessageFilePinDb.created > after_time)
+                | (
+                    (MessageFilePinDb.created == after_time)
+                    & (MessageFilePinDb.item_hash > after_hash)
+                )
+            )
+
+    if after_time is not None or cursor_mode:
+        if pagination:
+            select_stmt = select_stmt.limit(pagination + 1)
+    elif pagination and page > 1:
+        select_stmt = select_stmt.offset((page - 1) * pagination)
+        if pagination:
+            select_stmt = select_stmt.limit(pagination)
+    elif pagination:
+        select_stmt = select_stmt.limit(pagination)
+
     return session.execute(select_stmt).all()
 
 

src/aleph/db/accessors/messages.py

@@ -12,7 +12,6 @@
 from aleph.db.accessors.address_stats import escape_like_pattern
 from aleph.db.accessors.cost import delete_costs_for_message
 from aleph.db.models.message_counts import MessageCountsDb
-from aleph.toolkit.cursor import decode_cursor
 from aleph.toolkit.timestamp import coerce_to_datetime, utc_now
 from aleph.types.channel import Channel
 from aleph.types.db_session import DbSession
@@ -84,7 +83,9 @@ def make_matching_messages_query(
     page: int = 1,
     pagination: int = 20,
     include_confirmations: bool = False,
-    cursor: Optional[str] = None,
+    after_time: Optional[dt.datetime] = None,
+    after_hash: Optional[str] = None,
+    cursor_mode: bool = False,
     # TODO: remove once all filters are supported
     **kwargs,
 ) -> Select:
@@ -178,19 +179,17 @@ def make_matching_messages_query(
                 MessageDb.item_hash.asc(),
             )
 
-    # Cursor pagination (if cursor provided, ignore page)
-    if cursor:
-        time_val, hash_val = decode_cursor(cursor)
-        cursor_time = coerce_to_datetime(time_val)
+    # Cursor pagination (if cursor values provided, ignore page)
+    if after_time is not None:
         if sort_order == SortOrder.DESCENDING:
             select_stmt = select_stmt.where(
-                (MessageDb.time < cursor_time)
-                | ((MessageDb.time == cursor_time) & (MessageDb.item_hash > hash_val))
+                (MessageDb.time < after_time)
+                | ((MessageDb.time == after_time) & (MessageDb.item_hash > after_hash))
             )
         else:
             select_stmt = select_stmt.where(
-                (MessageDb.time > cursor_time)
-                | ((MessageDb.time == cursor_time) & (MessageDb.item_hash > hash_val))
+                (MessageDb.time > after_time)
+                | ((MessageDb.time == after_time) & (MessageDb.item_hash > after_hash))
             )
     elif page > 1:
         select_stmt = select_stmt.offset((page - 1) * pagination)
@@ -199,8 +198,9 @@ def make_matching_messages_query(
 
     # If pagination == 0, return all matching results
     if pagination:
-        # Fetch +1 for has_more detection when using cursor
-        select_stmt = select_stmt.limit(pagination + 1 if cursor else pagination)
+        select_stmt = select_stmt.limit(
+            pagination + 1 if after_time is not None or cursor_mode else pagination
+        )
 
     return select_stmt
 
@@ -309,6 +309,9 @@ def get_message_stats_by_address(
     sort_order: SortOrder = SortOrder.DESCENDING,
     page: int = 1,
     pagination: int = 0,
+    after_sort_value: Optional[Any] = None,
+    after_address: Optional[str] = None,
+    cursor_mode: bool = False,
 ) -> Sequence[Any]:
     """
     Get message stats for user addresses using the message_counts table.
@@ -388,7 +391,28 @@ def get_message_stats_by_address(
     else:
         stmt = stmt.order_by(sort_column.desc(), subquery.c.address.asc())
 
-    if pagination:
+    if after_sort_value is not None:
+        if sort_order == SortOrder.DESCENDING:
+            stmt = stmt.where(
+                (sort_column < after_sort_value)
+                | (
+                    (sort_column == after_sort_value)
+                    & (subquery.c.address > after_address)
+                )
+            )
+        else:
+            stmt = stmt.where(
+                (sort_column > after_sort_value)
+                | (
+                    (sort_column == after_sort_value)
+                    & (subquery.c.address > after_address)
+                )
+            )
+
+    if after_sort_value is not None or cursor_mode:
+        if pagination:
+            stmt = stmt.limit(pagination + 1)
+    elif pagination:
         stmt = stmt.limit(pagination).offset((page - 1) * pagination)
 
     return session.execute(stmt).all()