feat: add filters and sorting to /credit_history endpoint (#1099)

* feat: add filters and sorting to /credit_history endpoint

Add has_expiration boolean filter, exclude_payment_method list filter,
and configurable sort_by/sort_order with cursor pagination support.

* fix: address PR review - cursor sort_order, nullable columns cleanup

- Add sort_order to cursor encoding/decoding to prevent incorrect
  pagination when switching between ASC/DESC directions
- Remove payment_method from _NULLABLE_SORT_COLUMNS (always has a value)
- Add comments explaining why NULL inclusion is correct in keyset pagination
- Add cursor pagination test verifying NULLs appear correctly across pages

Author: amalcaraz

src/aleph/db/accessors/balances.py

@@ -18,6 +18,7 @@
 )
 from aleph.toolkit.timestamp import timestamp_to_datetime, utc_now
 from aleph.types.db_session import DbSession
+from aleph.types.sort_order import SortByCreditHistory, SortOrder
 
 
 def _apply_credit_precision_multiplier(
@@ -869,6 +870,63 @@ def validate_credit_transfer_balance(
     return current_balance >= total_transfer_amount
 
 
+CREDIT_HISTORY_SORT_COLUMN_MAP = {
+    SortByCreditHistory.MESSAGE_TIMESTAMP: AlephCreditHistoryDb.message_timestamp,
+    SortByCreditHistory.EXPIRATION_DATE: AlephCreditHistoryDb.expiration_date,
+    SortByCreditHistory.PAYMENT_METHOD: AlephCreditHistoryDb.payment_method,
+    SortByCreditHistory.AMOUNT: AlephCreditHistoryDb.amount,
+    SortByCreditHistory.ORIGIN: AlephCreditHistoryDb.origin,
+    SortByCreditHistory.TX_HASH: AlephCreditHistoryDb.tx_hash,
+    SortByCreditHistory.PROVIDER: AlephCreditHistoryDb.provider,
+}
+
+# Columns that are nullable and need NULLS LAST handling
+_NULLABLE_SORT_COLUMNS = {
+    SortByCreditHistory.EXPIRATION_DATE,
+    SortByCreditHistory.ORIGIN,
+    SortByCreditHistory.TX_HASH,
+    SortByCreditHistory.PROVIDER,
+}
+
+
+def _apply_credit_history_filters(
+    query: Select,
+    tx_hash: Optional[str] = None,
+    token: Optional[str] = None,
+    chain: Optional[str] = None,
+    provider: Optional[str] = None,
+    origin: Optional[str] = None,
+    origin_ref: Optional[str] = None,
+    payment_method: Optional[str] = None,
+    has_expiration: Optional[bool] = None,
+    exclude_payment_method: Optional[List[str]] = None,
+) -> Select:
+    """Apply common filters to a credit history query."""
+    if tx_hash is not None:
+        query = query.where(AlephCreditHistoryDb.tx_hash == tx_hash)
+    if token is not None:
+        query = query.where(AlephCreditHistoryDb.token == token)
+    if chain is not None:
+        query = query.where(AlephCreditHistoryDb.chain == chain)
+    if provider is not None:
+        query = query.where(AlephCreditHistoryDb.provider == provider)
+    if origin is not None:
+        query = query.where(AlephCreditHistoryDb.origin == origin)
+    if origin_ref is not None:
+        query = query.where(AlephCreditHistoryDb.origin_ref == origin_ref)
+    if payment_method is not None:
+        query = query.where(AlephCreditHistoryDb.payment_method == payment_method)
+    if has_expiration is True:
+        query = query.where(AlephCreditHistoryDb.expiration_date.isnot(None))
+    elif has_expiration is False:
+        query = query.where(AlephCreditHistoryDb.expiration_date.is_(None))
+    if exclude_payment_method:
+        query = query.where(
+            AlephCreditHistoryDb.payment_method.notin_(exclude_payment_method)
+        )
+    return query
+
+
 def get_address_credit_history(
     session: DbSession,
     address: str,
@@ -881,75 +939,122 @@ 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,
+    has_expiration: Optional[bool] = None,
+    exclude_payment_method: Optional[List[str]] = None,
+    sort_by: SortByCreditHistory = SortByCreditHistory.MESSAGE_TIMESTAMP,
+    sort_order: SortOrder = SortOrder.DESCENDING,
+    after_sort_value: Optional[Any] = 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.
+    Get paginated credit history entries for a specific address.
 
-    Args:
-        session: Database session
-        address: Address to get credit history for
-        page: Page number (starts at 1)
-        pagination: Number of entries per page (0 for all entries)
-        tx_hash: Filter by transaction hash
-        token: Filter by token
-        chain: Filter by chain
-        provider: Filter by provider
-        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
+    Supports dynamic sorting and cursor-based or page-based pagination.
     """
-    query = (
-        select(AlephCreditHistoryDb)
-        .where(AlephCreditHistoryDb.address == address)
-        .order_by(
-            AlephCreditHistoryDb.message_timestamp.desc(),
+    query = select(AlephCreditHistoryDb).where(AlephCreditHistoryDb.address == address)
+
+    # Dynamic ordering
+    primary_col = CREDIT_HISTORY_SORT_COLUMN_MAP[sort_by]
+    is_desc = sort_order == SortOrder.DESCENDING
+
+    if sort_by in _NULLABLE_SORT_COLUMNS:
+        if is_desc:
+            order_primary = primary_col.desc().nullslast()
+        else:
+            order_primary = primary_col.asc().nullslast()
+    else:
+        order_primary = primary_col.desc() if is_desc else primary_col.asc()
+
+    # Tiebreakers for stable pagination
+    if is_desc:
+        query = query.order_by(
+            order_primary,
             AlephCreditHistoryDb.credit_ref.desc(),
             AlephCreditHistoryDb.credit_index.desc(),
         )
-    )
+    else:
+        query = query.order_by(
+            order_primary,
+            AlephCreditHistoryDb.credit_ref.asc(),
+            AlephCreditHistoryDb.credit_index.asc(),
+        )
 
     # Apply filters
-    if tx_hash is not None:
-        query = query.where(AlephCreditHistoryDb.tx_hash == tx_hash)
-    if token is not None:
-        query = query.where(AlephCreditHistoryDb.token == token)
-    if chain is not None:
-        query = query.where(AlephCreditHistoryDb.chain == chain)
-    if provider is not None:
-        query = query.where(AlephCreditHistoryDb.provider == provider)
-    if origin is not None:
-        query = query.where(AlephCreditHistoryDb.origin == origin)
-    if origin_ref is not None:
-        query = query.where(AlephCreditHistoryDb.origin_ref == origin_ref)
-    if payment_method is not None:
-        query = query.where(AlephCreditHistoryDb.payment_method == payment_method)
+    query = _apply_credit_history_filters(
+        query,
+        tx_hash=tx_hash,
+        token=token,
+        chain=chain,
+        provider=provider,
+        origin=origin,
+        origin_ref=origin_ref,
+        payment_method=payment_method,
+        has_expiration=has_expiration,
+        exclude_payment_method=exclude_payment_method,
+    )
 
-    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)
+    # Cursor-based keyset pagination
+    if after_credit_ref is not None:
+        if after_sort_value is None and sort_by in _NULLABLE_SORT_COLUMNS:
+            # Last entry had NULL sort value — only compare tiebreakers within NULL group
+            if is_desc:
+                query = query.where(
+                    (primary_col.is_(None))
+                    & (
+                        (AlephCreditHistoryDb.credit_ref < after_credit_ref)
+                        | (
+                            (AlephCreditHistoryDb.credit_ref == after_credit_ref)
+                            & (AlephCreditHistoryDb.credit_index < after_credit_index)
+                        )
                     )
                 )
+            else:
+                query = query.where(
+                    (primary_col.is_(None))
+                    & (
+                        (AlephCreditHistoryDb.credit_ref > after_credit_ref)
+                        | (
+                            (AlephCreditHistoryDb.credit_ref == after_credit_ref)
+                            & (AlephCreditHistoryDb.credit_index > after_credit_index)
+                        )
+                    )
+                )
+        elif is_desc:
+            query = query.where(
+                (primary_col < after_sort_value)
+                | (
+                    (primary_col == after_sort_value)
+                    & (
+                        (AlephCreditHistoryDb.credit_ref < after_credit_ref)
+                        | (
+                            (AlephCreditHistoryDb.credit_ref == after_credit_ref)
+                            & (AlephCreditHistoryDb.credit_index < after_credit_index)
+                        )
+                    )
+                )
+                # NULLs come after all non-NULLs due to NULLS LAST ordering
+                | (primary_col.is_(None))
+            )
+        else:
+            query = query.where(
+                (primary_col > after_sort_value)
+                | (
+                    (primary_col == after_sort_value)
+                    & (
+                        (AlephCreditHistoryDb.credit_ref > after_credit_ref)
+                        | (
+                            (AlephCreditHistoryDb.credit_ref == after_credit_ref)
+                            & (AlephCreditHistoryDb.credit_index > after_credit_index)
+                        )
+                    )
+                )
+                # NULLs come after all non-NULLs due to NULLS LAST ordering
+                | (primary_col.is_(None))
             )
-        )
 
-    if after_time is not None or cursor_mode:
+    if after_credit_ref is not None or cursor_mode:
         if pagination > 0:
             query = query.limit(pagination + 1)
     elif pagination > 0:
@@ -968,43 +1073,28 @@ def count_address_credit_history(
     origin: Optional[str] = None,
     origin_ref: Optional[str] = None,
     payment_method: Optional[str] = None,
+    has_expiration: Optional[bool] = None,
+    exclude_payment_method: Optional[List[str]] = None,
 ) -> int:
     """
     Count total credit history entries for a specific address with optional filters.
-
-    Args:
-        session: Database session
-        address: Address to count credit history for
-        tx_hash: Filter by transaction hash
-        token: Filter by token
-        chain: Filter by chain
-        provider: Filter by provider
-        origin: Filter by origin
-        origin_ref: Filter by origin reference
-        payment_method: Filter by payment method
-
-    Returns:
-        Total number of credit history entries for the address matching the filters
     """
     query = select(func.count(AlephCreditHistoryDb.credit_ref)).where(
         AlephCreditHistoryDb.address == address
     )
 
-    # Apply filters
-    if tx_hash is not None:
-        query = query.where(AlephCreditHistoryDb.tx_hash == tx_hash)
-    if token is not None:
-        query = query.where(AlephCreditHistoryDb.token == token)
-    if chain is not None:
-        query = query.where(AlephCreditHistoryDb.chain == chain)
-    if provider is not None:
-        query = query.where(AlephCreditHistoryDb.provider == provider)
-    if origin is not None:
-        query = query.where(AlephCreditHistoryDb.origin == origin)
-    if origin_ref is not None:
-        query = query.where(AlephCreditHistoryDb.origin_ref == origin_ref)
-    if payment_method is not None:
-        query = query.where(AlephCreditHistoryDb.payment_method == payment_method)
+    query = _apply_credit_history_filters(
+        query,
+        tx_hash=tx_hash,
+        token=token,
+        chain=chain,
+        provider=provider,
+        origin=origin,
+        origin_ref=origin_ref,
+        payment_method=payment_method,
+        has_expiration=has_expiration,
+        exclude_payment_method=exclude_payment_method,
+    )
 
     return session.execute(query).scalar_one()
 

src/aleph/schemas/api/accounts.py

@@ -7,7 +7,7 @@
 
 from aleph.schemas.messages_query_params import DEFAULT_PAGE, LIST_FIELD_SEPARATOR
 from aleph.types.files import FileType
-from aleph.types.sort_order import SortOrder
+from aleph.types.sort_order import SortByCreditHistory, SortOrder
 
 
 class GetAccountQueryParams(BaseModel):
@@ -158,6 +158,30 @@ class GetAccountCreditHistoryQueryParams(BaseModel):
     payment_method: Optional[str] = Field(
         default=None, description="Filter by payment method"
     )
+    has_expiration: Optional[bool] = Field(
+        default=None,
+        description="Filter by presence of expiration_date. "
+        "true: only entries with an expiration date, "
+        "false: only entries without an expiration date.",
+    )
+    exclude_payment_method: Optional[List[str]] = Field(
+        default=None,
+        description="Exclude entries matching these payment methods (comma-separated).",
+    )
+    sort_by: SortByCreditHistory = Field(
+        default=SortByCreditHistory.MESSAGE_TIMESTAMP,
+        description="Field to sort by.",
+    )
+    sort_order: SortOrder = Field(
+        default=SortOrder.DESCENDING,
+        description="Sort direction: 1 (ASC) or -1 (DESC).",
+    )
+
+    @field_validator("exclude_payment_method", mode="before")
+    def split_exclude_payment_method(cls, v):
+        if isinstance(v, str):
+            return v.split(LIST_FIELD_SEPARATOR)
+        return v
 
 
 class CreditHistoryResponseItem(BaseModel):

src/aleph/toolkit/cursor.py

@@ -90,6 +90,47 @@ def decode_credit_history_cursor(cursor: str) -> Tuple[dt.datetime, str, int]:
         raise ValueError("Invalid cursor: missing required fields")
 
 
+def encode_credit_history_sort_cursor(
+    sort_by: str,
+    sort_value: Any,
+    sort_order: int,
+    credit_ref: str,
+    credit_index: int,
+) -> str:
+    """Encode a credit history cursor with sort field and order info."""
+    value = (
+        sort_value.isoformat() if isinstance(sort_value, dt.datetime) else sort_value
+    )
+    return encode_cursor(
+        {"s": sort_by, "v": value, "o": sort_order, "r": credit_ref, "i": credit_index}
+    )
+
+
+def decode_credit_history_sort_cursor(
+    cursor: str,
+) -> Tuple[str, Any, int, str, int]:
+    """Decode a credit history sort cursor.
+
+    Returns (sort_by, sort_value, sort_order, credit_ref, credit_index).
+    For backward compat, if 's' key is missing, assumes 'message_timestamp' sort
+    with DESC order and uses the 't' key as the sort value.
+    """
+    try:
+        d = decode_cursor(cursor)
+        if "s" in d:
+            return (
+                str(d["s"]),
+                d["v"],
+                int(d["o"]),
+                str(d["r"]),
+                int(d["i"]),
+            )
+        # Backward compat: old cursor format with (t, r, i)
+        return "message_timestamp", d["t"], -1, str(d["r"]), int(d["i"])
+    except KeyError:
+        raise ValueError("Invalid cursor: missing required fields")
+
+
 def encode_address_stats_cursor(sort_value: Any, address: str) -> str:
     """Encode an address stats cursor: (sort_value, address)."""
     return encode_cursor({"v": sort_value, "a": address})