Harden payment observability and polling

Dockerfile

@@ -21,8 +21,9 @@ COPY --from=builder /app/src /app/src
 ENV PATH="/app/.venv/bin:$PATH" \
     PYTHONUNBUFFERED=1 \
     PYTHONDONTWRITEBYTECODE=1 \
-    PYTHONPATH=/app
+    PYTHONPATH=/app \
+    PROMETHEUS_MULTIPROC_DIR=/tmp/prometheus_multiproc
 
 EXPOSE 8000
 
-CMD ["sh", "-c", "granian --interface asgi ${APP_MODULE:-src.main:app} --host ${APP_HOST:-0.0.0.0} --port ${APP_PORT:-8000} --workers ${APP_WORKERS:-2} --loop uvloop"]
+CMD ["sh", "-c", "mkdir -p "$PROMETHEUS_MULTIPROC_DIR" && rm -f "$PROMETHEUS_MULTIPROC_DIR"/* && granian --interface asgi ${APP_MODULE:-src.main:app} --host ${APP_HOST:-0.0.0.0} --port ${APP_PORT:-8000} --workers ${APP_WORKERS:-2} --loop uvloop"]

pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "hvac==2.4.0",
     "itsdangerous==2.2.0",
     "orjson==3.11.7",
+    "prometheus-client==0.23.1",
     "pydantic-settings==2.12.0",
     "python-jose==3.5.0",
     "python-ulid==3.1.0",

src/application/commands/payment_read_commands.py

@@ -25,6 +25,15 @@ class ClientOperationResult:
     withdrawal: SbpWithdrawalEntity | None = None
 
 
+def _parse_cursor(cursor: str | None) -> datetime | None:
+    if not cursor:
+        return None
+    parsed = datetime.fromisoformat(cursor.replace('Z','+00:00'))
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
 class GetPaymentConfigCommand:
     def __init__(self, *, payment_quote_service: PaymentQuoteService, logger: ILogger):
         self._payment_quote_service = payment_quote_service
@@ -68,11 +77,12 @@ class ListOrdersCommand:
 
 
     @transactional
-    async def __call__(self, *, user_id: str, limit: int, offset: int) -> list[OrderPaymentResult]:
+    async def __call__(self, *, user_id: str, limit: int, offset: int, cursor: str | None = None) -> list[OrderPaymentResult]:
         orders = await self._unit_of_work.order_repository.list_by_user_id(
             user_id=user_id,
             limit=limit,
             offset=offset,
+            cursor_created_at=_parse_cursor(cursor),
         )
         items: list[OrderPaymentResult] = []
         for order in orders:
@@ -80,7 +90,7 @@ class ListOrdersCommand:
                 continue
             payment = await self._unit_of_work.payment_repository.get_by_order_id(order.id)
             items.append(OrderPaymentResult(order=order,payment=payment))
-        self._logger.info({'event':'orders_list_requested','user_id':user_id,'limit':limit,'offset':offset})
+        self._logger.info({'event':'orders_list_requested','user_id':user_id,'limit':limit,'offset':offset,'cursor':cursor})
         return items
 
 
@@ -108,17 +118,20 @@ class ListClientOperationsCommand:
 
 
     @transactional
-    async def __call__(self, *, user_id: str, limit: int, offset: int) -> list[ClientOperationResult]:
-        fetch_limit=limit+offset
+    async def __call__(self, *, user_id: str, limit: int, offset: int, cursor: str | None = None) -> list[ClientOperationResult]:
+        cursor_created_at=_parse_cursor(cursor)
+        fetch_limit=limit if cursor_created_at is not None else limit+offset
         payments=await self._unit_of_work.payment_repository.list_by_user_id(
             user_id=user_id,
             limit=fetch_limit,
             offset=0,
+            cursor_created_at=cursor_created_at,
         )
         withdrawals=await self._unit_of_work.sbp_withdrawal_repository.list_by_user_id(
             user_id=user_id,
             limit=fetch_limit,
             offset=0,
+            cursor_created_at=cursor_created_at,
         )
         min_dt=datetime.min.replace(tzinfo=timezone.utc)
         items=[
@@ -130,8 +143,8 @@ class ListClientOperationsCommand:
             for withdrawal in withdrawals
         )
         items.sort(key=lambda item:item.created_at,reverse=True)
-        self._logger.info({'event':'client_operations_list_requested','user_id':user_id,'limit':limit,'offset':offset})
-        return items[offset:offset+limit]
+        self._logger.info({'event':'client_operations_list_requested','user_id':user_id,'limit':limit,'offset':offset,'cursor':cursor})
+        return items[:limit] if cursor_created_at is not None else items[offset:offset+limit]
 
 
 class GetPaymentCommand:

src/infrastructure/database/repositories/order_repository.py

@@ -94,14 +94,17 @@ class OrderRepository(IOrderRepository):
         return self._to_entity(model)
 
 
-    async def list_by_user_id(self,*,user_id: str,limit: int,offset: int) -> list[OrderEntity]:
+    async def list_by_user_id(self,*,user_id: str,limit: int,offset: int,cursor_created_at=None) -> list[OrderEntity]:
         stmt=(
             select(Order)
             .where(Order.user_id==user_id)
             .order_by(desc(Order.created_at))
             .limit(limit)
-            .offset(offset)
         )
+        if cursor_created_at is not None:
+            stmt=stmt.where(Order.created_at < cursor_created_at)
+        else:
+            stmt=stmt.offset(offset)
         result=await self._session.scalars(stmt)
         return [self._to_entity(model) for model in result.all()]
 

src/infrastructure/database/repositories/payment_repository.py

@@ -38,28 +38,28 @@ class PaymentRepository(IPaymentRepository):
 
 
     async def create_completed(self,*,user_id:str,order_id:str,itpay_payment_id:str,itpay_paid_amount:str|None,transaction_id:str|None,paid_at:str|None,expired_date:str|None) -> bool:
-        stmt=select(Payment).where(Payment.order_id==order_id)
-        existing=await self._session.scalar(stmt)
-        if existing is not None:
-            return False
         paid_at_dt=datetime.fromisoformat(paid_at.replace('Z','+00:00')) if paid_at else None
         expired_dt=datetime.fromisoformat(expired_date.replace('Z','+00:00')) if expired_date else None
         paid_amount_dec=Decimal(str(itpay_paid_amount)) if itpay_paid_amount is not None else None
-        model=Payment(
-            user_id=user_id,
-            order_id=order_id,
-            status=PaymentStatus.MONEY_ACCEPTED,
-            receipt_cloudekassir_id=None,
-            receipt_cloudekassir_link=None,
-            itpay_payment_id=itpay_payment_id,
-            itpay_paid_amount=paid_amount_dec,
-            transaction_id=transaction_id,
-            paid_at=paid_at_dt,
-            expired_date=expired_dt,
+        stmt=(
+            insert(Payment)
+            .values(
+                user_id=user_id,
+                order_id=order_id,
+                status=PaymentStatus.MONEY_ACCEPTED,
+                receipt_cloudekassir_id=None,
+                receipt_cloudekassir_link=None,
+                itpay_payment_id=itpay_payment_id,
+                itpay_paid_amount=paid_amount_dec,
+                transaction_id=transaction_id,
+                paid_at=paid_at_dt,
+                expired_date=expired_dt,
+            )
+            .on_conflict_do_nothing(index_elements=[Payment.order_id])
         )
-        self._session.add(model)
+        result=await self._session.execute(stmt)
         await self._session.flush()
-        return True
+        return bool(result.rowcount)
 
 
     async def update_crypto_transfer_completed(self,*,order_id:str,web3_transaction_hash:str|None) -> None:
@@ -111,14 +111,17 @@ class PaymentRepository(IPaymentRepository):
         return self._to_entity(model)
 
 
-    async def list_by_user_id(self,*,user_id:str,limit:int,offset:int) -> list[PaymentEntity]:
+    async def list_by_user_id(self,*,user_id:str,limit:int,offset:int,cursor_created_at=None) -> list[PaymentEntity]:
         stmt=(
             select(Payment)
             .where(Payment.user_id==user_id)
             .order_by(desc(Payment.created_at))
             .limit(limit)
-            .offset(offset)
         )
+        if cursor_created_at is not None:
+            stmt=stmt.where(Payment.created_at < cursor_created_at)
+        else:
+            stmt=stmt.offset(offset)
         result=await self._session.scalars(stmt)
         return [self._to_entity(model) for model in result.all()]
 

src/infrastructure/database/repositories/sbp_withdrawal_repository.py

@@ -84,14 +84,17 @@ class SbpWithdrawalRepository(ISbpWithdrawalRepository):
         return self._to_entity(model)
 
 
-    async def list_by_user_id(self,*,user_id: str,limit: int,offset: int) -> list[SbpWithdrawalEntity]:
+    async def list_by_user_id(self,*,user_id: str,limit: int,offset: int,cursor_created_at=None) -> list[SbpWithdrawalEntity]:
         stmt=(
             select(SbpWithdrawal)
             .where(SbpWithdrawal.user_id==user_id)
             .order_by(desc(SbpWithdrawal.created_at))
             .limit(limit)
-            .offset(offset)
         )
+        if cursor_created_at is not None:
+            stmt=stmt.where(SbpWithdrawal.created_at < cursor_created_at)
+        else:
+            stmt=stmt.offset(offset)
         result=await self._session.scalars(stmt)
         return [self._to_entity(model) for model in result.all()]
 

src/infrastructure/metrics.py

@@ -1,52 +1,40 @@
 from __future__ import annotations
 
+import asyncio
+import os
 import time
-from collections import defaultdict
+from collections.abc import Awaitable, Callable
 from datetime import timedelta
-from typing import Iterable
+from typing import Any
 
 from fastapi import Request, Response
+from prometheus_client import CollectorRegistry, Counter, Histogram, CONTENT_TYPE_LATEST, generate_latest, multiprocess
 from sqlalchemy import func, select
 from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
 from starlette.responses import Response as StarletteResponse
 
 from src.infrastructure.database.context import async_session_maker
+from src.infrastructure.database.models import Order, Payment
 
-
-SERVICE_NAME = "pay"
-WINDOWS: tuple[tuple[str, timedelta | None], ...] = (
-    ("all", None),
-    ("30d", timedelta(days=30)),
-    ("7d", timedelta(days=7)),
-    ("24h", timedelta(days=1)),
-    ("1h", timedelta(hours=1)),
-)
+SERVICE_NAME = "payment"
+BUSINESS_CACHE_TTL_SECONDS = 30
 LATENCY_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0)
+WINDOWS = (("all", None), ("30d", timedelta(days=30)), ("7d", timedelta(days=7)), ("24h", timedelta(days=1)), ("1h", timedelta(hours=1)))
 
-_http_requests: dict[tuple[str, str, str], int] = defaultdict(int)
-_http_latency_sum: dict[tuple[str, str], float] = defaultdict(float)
-_http_latency_count: dict[tuple[str, str], int] = defaultdict(int)
-_http_latency_buckets: dict[tuple[str, str, float], int] = defaultdict(int)
-
-
-def _labels(labels: dict[str, str]) -> str:
-    return ",".join(f'{key}="{value}"' for key, value in labels.items())
-
-
-def _sample(name: str, value: int | float, labels: dict[str, str] | None = None) -> str:
-    return f"{name}{{{_labels(labels)}}} {value}" if labels else f"{name} {value}"
+HTTP_REQUESTS = Counter("payment_http_requests_total", "Total HTTP requests handled by payment service.", ("service", "method", "route", "status_code"))
+HTTP_LATENCY = Histogram("payment_http_request_duration_seconds", "HTTP request duration in seconds for payment service.", ("service", "method", "route"), buckets=LATENCY_BUCKETS)
 
 
 def _route_template(request: Request) -> str:
     route = request.scope.get("route")
-    return str(getattr(route, "path", None) or request.url.path)
+    path = getattr(route, "path", None)
+    return str(path) if path else "__unmatched__"
 
 
 class PrometheusMetricsMiddleware(BaseHTTPMiddleware):
     async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> StarletteResponse:
         if request.url.path == "/metrics":
             return await call_next(request)
-
         started = time.perf_counter()
         status_code = "500"
         try:
@@ -56,87 +44,56 @@ class PrometheusMetricsMiddleware(BaseHTTPMiddleware):
         finally:
             elapsed = time.perf_counter() - started
             route = _route_template(request)
-            method = request.method
-            _http_requests[(method, route, status_code)] += 1
-            _http_latency_sum[(method, route)] += elapsed
-            _http_latency_count[(method, route)] += 1
-            for bucket in LATENCY_BUCKETS:
-                if elapsed <= bucket:
-                    _http_latency_buckets[(method, route, bucket)] += 1
+            HTTP_REQUESTS.labels(SERVICE_NAME, request.method, route, status_code).inc()
+            HTTP_LATENCY.labels(SERVICE_NAME, request.method, route).observe(elapsed)
 
 
-async def _count(statement) -> int:
-    async with async_session_maker() as session:
-        result = await session.execute(statement)
-        return int(result.scalar_one() or 0)
+class CachedBusinessMetrics:
+    def __init__(self, ttl_seconds: int, collector: Callable[[], Awaitable[list[str]]]) -> None:
+        self._ttl_seconds = ttl_seconds
+        self._collector = collector
+        self._expires_at = 0.0
+        self._lines: list[str] = []
+        self._lock = asyncio.Lock()
+
+    async def get(self) -> list[str]:
+        now = time.monotonic()
+        if now < self._expires_at:
+            return self._lines
+        async with self._lock:
+            now = time.monotonic()
+            if now < self._expires_at:
+                return self._lines
+            self._lines = await self._collector()
+            self._expires_at = now + self._ttl_seconds
+            return self._lines
 
 
-async def _rows(statement):
-    async with async_session_maker() as session:
-        result = await session.execute(statement)
-        return result.all()
+def _labels(labels: dict[str, str]) -> str:
+    escaped = {key: value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") for key, value in labels.items()}
+    return ",".join(f'{key}="{value}"' for key, value in escaped.items())
 
 
-def _http_metrics(prefix: str) -> Iterable[str]:
-    yield f"# TYPE {prefix}_http_requests_total counter"
-    for (method, route, status_code), value in sorted(_http_requests.items()):
-        yield _sample(
-            f"{prefix}_http_requests_total",
-            value,
-            {"service": SERVICE_NAME, "method": method, "route": route, "status_code": status_code},
-        )
-
-    yield f"# TYPE {prefix}_http_request_duration_seconds histogram"
-    for method, route in sorted(_http_latency_count):
-        for bucket in LATENCY_BUCKETS:
-            yield _sample(
-                f"{prefix}_http_request_duration_seconds_bucket",
-                _http_latency_buckets.get((method, route, bucket), 0),
-                {"service": SERVICE_NAME, "method": method, "route": route, "le": str(bucket)},
-            )
-        yield _sample(
-            f"{prefix}_http_request_duration_seconds_bucket",
-            _http_latency_count[(method, route)],
-            {"service": SERVICE_NAME, "method": method, "route": route, "le": "+Inf"},
-        )
-        yield _sample(
-            f"{prefix}_http_request_duration_seconds_sum",
-            _http_latency_sum[(method, route)],
-            {"service": SERVICE_NAME, "method": method, "route": route},
-        )
-        yield _sample(
-            f"{prefix}_http_request_duration_seconds_count",
-            _http_latency_count[(method, route)],
-            {"service": SERVICE_NAME, "method": method, "route": route},
-        )
-
-from src.infrastructure.database.models.order import Order
-from src.infrastructure.database.models.payment import Payment
+def _sample(name: str, value: int | float, labels: dict[str, str] | None = None) -> str:
+    return f"{name}{{{_labels(labels)}}} {value}" if labels else f"{name} {value}"
 
 
-async def _business_metrics() -> list[str]:
-    lines: list[str] = []
-    for model, metric_prefix in ((Order, "payment_orders"), (Payment, "payments")):
-        total = await _count(select(func.count()).select_from(model))
-        lines.append(_sample(f"{metric_prefix}_total", total, {"service": SERVICE_NAME}))
-        for status, count in await _rows(select(model.status, func.count()).group_by(model.status)):
-            lines.append(_sample(f"{metric_prefix}_by_status_total", int(count), {"service": SERVICE_NAME, "status": getattr(status, "value", str(status))}))
-        for window_name, delta in WINDOWS:
-            conditions = []
-            if delta is not None:
-                conditions.append(model.created_at >= func.now() - delta)
-            created = await _count(select(func.count()).select_from(model).where(*conditions))
-            lines.append(_sample(f"{metric_prefix}_created_total", created, {"service": SERVICE_NAME, "window": window_name}))
-
-    paid_total = await _count(select(func.count()).select_from(Payment).where(Payment.paid_at.is_not(None)))
-    failed_total = await _count(select(func.count()).select_from(Payment).where(Payment.status.in_(["web3_hash_error", "web3_balance_problem", "receipt_error"])))
-    lines.append(_sample("payments_succeeded_total", paid_total, {"service": SERVICE_NAME}))
-    lines.append(_sample("payments_failed_total", failed_total, {"service": SERVICE_NAME}))
-    return lines
+async def _scalar(session: Any, statement: Any) -> int:
+    result = await session.execute(statement)
+    return int(result.scalar_one() or 0)
 
 
-async def metrics_response() -> Response:
-    lines = [
+async def _rows(session: Any, statement: Any) -> list[Any]:
+    result = await session.execute(statement)
+    return list(result.all())
+
+
+def _status_value(status: Any) -> str:
+    return getattr(status, "value", str(status))
+
+
+async def _collect_business_metrics() -> list[str]:
+    lines: list[str] = [
         "# TYPE payment_orders_total gauge",
         "# TYPE payment_orders_by_status_total gauge",
         "# TYPE payment_orders_created_total gauge",
@@ -146,10 +103,42 @@ async def metrics_response() -> Response:
         "# TYPE payments_succeeded_total gauge",
         "# TYPE payments_failed_total gauge",
     ]
+    async with async_session_maker() as session:
+        for model, metric_prefix in ((Order, "payment_orders"), (Payment, "payments")):
+            total = await _scalar(session, select(func.count()).select_from(model))
+            lines.append(_sample(f"{metric_prefix}_total", total, {"service": SERVICE_NAME}))
+            for status, count in await _rows(session, select(model.status, func.count()).group_by(model.status)):
+                lines.append(_sample(f"{metric_prefix}_by_status_total", int(count), {"service": SERVICE_NAME, "status": _status_value(status)}))
+            for window_name, delta in WINDOWS:
+                conditions = []
+                if delta is not None:
+                    conditions.append(model.created_at >= func.now() - delta)
+                created = await _scalar(session, select(func.count()).select_from(model).where(*conditions))
+                lines.append(_sample(f"{metric_prefix}_created_total", created, {"service": SERVICE_NAME, "window": window_name}))
+        paid_total = await _scalar(session, select(func.count()).select_from(Payment).where(Payment.paid_at.is_not(None)))
+        failed_total = await _scalar(session, select(func.count()).select_from(Payment).where(Payment.status.in_(["web3_hash_error", "web3_balance_problem", "receipt_error"])))
+        lines.append(_sample("payments_succeeded_total", paid_total, {"service": SERVICE_NAME}))
+        lines.append(_sample("payments_failed_total", failed_total, {"service": SERVICE_NAME}))
+    lines.append(_sample("payment_metrics_scrape_success", 1, {"service": SERVICE_NAME}))
+    return lines
+
+
+_business_metrics_cache = CachedBusinessMetrics(BUSINESS_CACHE_TTL_SECONDS, _collect_business_metrics)
+
+
+def _client_metrics() -> bytes:
+    if os.environ.get("PROMETHEUS_MULTIPROC_DIR"):
+        registry = CollectorRegistry()
+        multiprocess.MultiProcessCollector(registry)
+        return generate_latest(registry)
+    return generate_latest()
+
+
+async def metrics_response() -> Response:
+    lines: list[str] = []
     try:
-        lines.extend(await _business_metrics())
-        lines.append(_sample("pay_metrics_scrape_success", 1, {"service": SERVICE_NAME}))
+        lines.extend(await _business_metrics_cache.get())
     except Exception:
-        lines.append(_sample("pay_metrics_scrape_success", 0, {"service": SERVICE_NAME}))
-    lines.extend(_http_metrics("pay"))
-    return Response("\n".join(lines) + "\n", media_type="text/plain; version=0.0.4; charset=utf-8")
+        lines.append(_sample("payment_metrics_scrape_success", 0, {"service": SERVICE_NAME}))
+    body = _client_metrics().decode("utf-8") + "\n".join(lines) + "\n"
+    return Response(body, media_type=CONTENT_TYPE_LATEST)

src/presentation/routing/order.py

@@ -10,7 +10,7 @@ from src.application.commands import ClientOperationResult,CreateOrderCommand,Cr
 from src.application.contracts import IJwtService,ILogger
 from src.application.domain.dto import AccessTokenPayload,AuthContext
 from src.application.domain.entities import OrderEntity,PaymentEntity,SbpWithdrawalEntity
-from src.application.domain.enums import OrderStatus
+from src.application.domain.enums import OrderStatus, PaymentStatus
 from src.application.domain.exceptions import ApplicationException,ConflictException
 from src.application.services import PaymentQuote
 from src.infrastructure.context_vars import trace_id_var
@@ -21,6 +21,18 @@ from src.presentation.dependencies.security import get_jwt_service
 from src.presentation.schemas.order import ClientOperationResponse,ClientOperationsResponse,CreateOrder,CreateOrderResponse,ErrorResponse,OrderDetailResponse,OrderPaymentResponse,OrdersResponse,OrderStatusResponse,OrderWithPaymentResponse,PaymentConfigResponse,PaymentDetailResponse,PaymentQuoteResponse,PaymentResponse,SbpWithdrawalOperationResponse
 from src.presentation.schemas.itpay_payment_models import ItpayPaymentData
 
+ORDER_EVENTS_CACHE_TTL_SECONDS = 10
+ORDER_EVENTS_TERMINAL_CACHE_TTL_SECONDS = 300
+ORDER_EVENTS_IDLE_SLEEP_SECONDS = 2
+TERMINAL_ORDER_STATUSES = {OrderStatus.COMPLETED, OrderStatus.CANCELLED, OrderStatus.ERROR, OrderStatus.REJECTED}
+TERMINAL_PAYMENT_STATUSES = {
+    PaymentStatus.COMPLETED,
+    PaymentStatus.USDT_DELIVERED,
+    PaymentStatus.WEB3_HASH_ERROR,
+    PaymentStatus.WEB3_BALANCE_PROBLEM,
+    PaymentStatus.RECEIPT_ERROR,
+}
+
 order_router = APIRouter(prefix='/order', tags=['orders'])
 orders_router = APIRouter(tags=['orders'])
 payment_router = APIRouter(prefix='/payment', tags=['payments'])
@@ -170,6 +182,23 @@ def _order_status_response(order: OrderEntity, payment: PaymentEntity | None) ->
     )
 
 
+def _is_terminal_order_status(order: OrderEntity, payment: PaymentEntity | None) -> bool:
+    if order.status in TERMINAL_ORDER_STATUSES:
+        return True
+    return bool(payment and payment.status in TERMINAL_PAYMENT_STATUSES)
+
+
+def _is_terminal_status_payload(payload: dict) -> bool:
+    data = payload.get('data') if isinstance(payload, dict) else None
+    if not isinstance(data, dict):
+        return False
+    order_status = data.get('order_status')
+    payment_status = data.get('payment_status')
+    terminal_order_values = {status.value for status in TERMINAL_ORDER_STATUSES}
+    terminal_payment_values = {status.value for status in TERMINAL_PAYMENT_STATUSES}
+    return order_status in terminal_order_values or payment_status in terminal_payment_values
+
+
 def _extract_websocket_access_token(websocket: WebSocket) -> str | None:
     token = websocket.cookies.get('access_token')
     if token:
@@ -289,12 +318,14 @@ async def payment_list_orders(
         request: Request,
         limit: int = Query(default=20, ge=1, le=100),
         offset: int = Query(default=0, ge=0),
+        cursor: str | None = Query(default=None),
         auth: AuthContext = Depends(require_access_token),
         command: ListOrdersCommand = Depends(get_list_orders_command),
 ) -> OrdersResponse:
-    orders = await command(user_id=auth.user_id,limit=limit,offset=offset)
+    orders = await command(user_id=auth.user_id,limit=limit,offset=offset,cursor=cursor)
     items = [_order_with_payment_response(item.order,item.payment) for item in orders]
-    return OrdersResponse(status_code=200,orders=items,limit=limit,offset=offset)
+    next_cursor = items[-1].order.created_at if items and items[-1].order.created_at else None
+    return OrdersResponse(status_code=200,orders=items,limit=limit,offset=offset,next_cursor=next_cursor)
 
 
 @payment_router.get(
@@ -308,15 +339,17 @@ async def payment_list_payments(
         request: Request,
         limit: int = Query(default=20, ge=1, le=100),
         offset: int = Query(default=0, ge=0),
+        cursor: str | None = Query(default=None),
         auth: AuthContext = Depends(require_access_token),
         command: ListClientOperationsCommand = Depends(get_list_client_operations_command),
 ) -> ClientOperationsResponse:
-    operations = await command(user_id=auth.user_id,limit=limit,offset=offset)
+    operations = await command(user_id=auth.user_id,limit=limit,offset=offset,cursor=cursor)
     return ClientOperationsResponse(
         status_code=200,
         operations=[_client_operation_response(operation) for operation in operations],
         limit=limit,
         offset=offset,
+        next_cursor=operations[-1].created_at.isoformat() if operations else None,
     )
 
 
@@ -348,12 +381,14 @@ async def list_orders(
         request: Request,
         limit: int = Query(default=20, ge=1, le=100),
         offset: int = Query(default=0, ge=0),
+        cursor: str | None = Query(default=None),
         auth: AuthContext = Depends(require_access_token),
         command: ListOrdersCommand = Depends(get_list_orders_command),
 ) -> OrdersResponse:
-    orders = await command(user_id=auth.user_id,limit=limit,offset=offset)
+    orders = await command(user_id=auth.user_id,limit=limit,offset=offset,cursor=cursor)
     items = [_order_with_payment_response(item.order,item.payment) for item in orders]
-    return OrdersResponse(status_code=200,orders=items,limit=limit,offset=offset)
+    next_cursor = items[-1].order.created_at if items and items[-1].order.created_at else None
+    return OrdersResponse(status_code=200,orders=items,limit=limit,offset=offset,next_cursor=next_cursor)
 
 
 @payments_router.get(
@@ -367,15 +402,17 @@ async def list_payments(
         request: Request,
         limit: int = Query(default=20, ge=1, le=100),
         offset: int = Query(default=0, ge=0),
+        cursor: str | None = Query(default=None),
         auth: AuthContext = Depends(require_access_token),
         command: ListClientOperationsCommand = Depends(get_list_client_operations_command),
 ) -> ClientOperationsResponse:
-    operations = await command(user_id=auth.user_id,limit=limit,offset=offset)
+    operations = await command(user_id=auth.user_id,limit=limit,offset=offset,cursor=cursor)
     return ClientOperationsResponse(
         status_code=200,
         operations=[_client_operation_response(operation) for operation in operations],
         limit=limit,
         offset=offset,
+        next_cursor=operations[-1].created_at.isoformat() if operations else None,
     )
 
 
@@ -413,20 +450,40 @@ async def order_events(
             return
         await websocket.accept()
         logger.info({'event':'order_events_connected','order_id':order_id,'user_id':auth.user_id})
+        redis_cache = getattr(websocket.app.state, 'redis_cache', None)
+        cache_key = f'order_status:{auth.user_id}:{order_id}'
         last_payload: dict | None = None
         while True:
-            try:
-                result = await command(order_id=order_id,user_id=auth.user_id)
-            except ApplicationException as exception:
-                await websocket.send_json({'event':'order_events_error','detail':exception.message,'status_code':exception.status_code})
-                await websocket.close(code=1008)
-                return
-            status_payload = _order_status_response(result.order,result.payment).model_dump(mode='json')
-            payload = {'event':'order_status','data':status_payload}
+            payload: dict | None = None
+            if redis_cache is not None:
+                try:
+                    cached = await redis_cache.get(cache_key)
+                    if cached:
+                        payload = orjson.loads(cached)
+                except Exception as exception:
+                    logger.warning({'event':'order_events_cache_read_failed','order_id':order_id,'error':str(exception)})
+            if payload is None:
+                try:
+                    result = await command(order_id=order_id,user_id=auth.user_id)
+                except ApplicationException as exception:
+                    await websocket.send_json({'event':'order_events_error','detail':exception.message,'status_code':exception.status_code})
+                    await websocket.close(code=1008)
+                    return
+                status_payload = _order_status_response(result.order,result.payment).model_dump(mode='json')
+                payload = {'event':'order_status','data':status_payload}
+                if redis_cache is not None:
+                    ttl = ORDER_EVENTS_TERMINAL_CACHE_TTL_SECONDS if _is_terminal_order_status(result.order,result.payment) else ORDER_EVENTS_CACHE_TTL_SECONDS
+                    try:
+                        await redis_cache.set(cache_key,orjson.dumps(payload).decode(),ex=ttl)
+                    except Exception as exception:
+                        logger.warning({'event':'order_events_cache_write_failed','order_id':order_id,'error':str(exception)})
             if payload != last_payload:
                 await websocket.send_text(orjson.dumps(payload).decode())
                 last_payload = payload
-            await asyncio.sleep(2)
+            if _is_terminal_status_payload(payload):
+                await websocket.close(code=1000)
+                return
+            await asyncio.sleep(ORDER_EVENTS_IDLE_SLEEP_SECONDS)
     except WebSocketDisconnect:
         logger.info({'event':'order_events_disconnected','order_id':order_id})
     finally:

src/presentation/schemas/order.py

@@ -120,6 +120,7 @@ class OrdersResponse(BaseModel):
     orders: list[OrderWithPaymentResponse]
     limit: int
     offset: int
+    next_cursor: str | None = None
 
 
 class PaymentsResponse(BaseModel):
@@ -134,6 +135,7 @@ class ClientOperationsResponse(BaseModel):
     operations: list[ClientOperationResponse]
     limit: int
     offset: int
+    next_cursor: str | None = None
 
 
 class PaymentDetailResponse(BaseModel):

uv.lock

@@ -398,9 +398,7 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/65/8b/3669ad3b3f247a791b2b4aceb3aa5a31f5f6817bf547e4e1ff712338145a/greenlet-3.4.0-cp312-cp312-macosx_11_0_universal2.whl", hash = "sha256:1a54a921561dd9518d31d2d3db4d7f80e589083063ab4d3e2e950756ef809e1a", size = 286902, upload-time = "2026-04-08T15:52:12.138Z" },
     { url = "https://files.pythonhosted.org/packages/38/3e/3c0e19b82900873e2d8469b590a6c4b3dfd2b316d0591f1c26b38a4879a5/greenlet-3.4.0-cp312-cp312-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:16dec271460a9a2b154e3b1c2fa1050ce6280878430320e85e08c166772e3f97", size = 606099, upload-time = "2026-04-08T16:24:38.408Z" },
     { url = "https://files.pythonhosted.org/packages/b5/33/99fef65e7754fc76a4ed14794074c38c9ed3394a5bd129d7f61b705f3168/greenlet-3.4.0-cp312-cp312-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:90036ce224ed6fe75508c1907a77e4540176dcf0744473627785dd519c6f9996", size = 618837, upload-time = "2026-04-08T16:30:58.298Z" },
-    { url = "https://files.pythonhosted.org/packages/44/57/eae2cac10421feae6c0987e3dc106c6d86262b1cb379e171b017aba893a6/greenlet-3.4.0-cp312-cp312-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:6f0def07ec9a71d72315cf26c061aceee53b306c36ed38c35caba952ea1b319d", size = 624901, upload-time = "2026-04-08T16:40:38.981Z" },
     { url = "https://files.pythonhosted.org/packages/36/f7/229f3aed6948faa20e0616a0b8568da22e365ede6a54d7d369058b128afd/greenlet-3.4.0-cp312-cp312-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:a1c4f6b453006efb8310affb2d132832e9bbb4fc01ce6df6b70d810d38f1f6dc", size = 615062, upload-time = "2026-04-08T15:56:33.766Z" },
-    { url = "https://files.pythonhosted.org/packages/6a/8a/0e73c9b94f31d1cc257fe79a0eff621674141cdae7d6d00f40de378a1e42/greenlet-3.4.0-cp312-cp312-manylinux_2_39_riscv64.whl", hash = "sha256:0e1254cf0cbaa17b04320c3a78575f29f3c161ef38f59c977108f19ffddaf077", size = 423927, upload-time = "2026-04-08T16:43:05.293Z" },
     { url = "https://files.pythonhosted.org/packages/08/97/d988180011aa40135c46cd0d0cf01dd97f7162bae14139b4a3ef54889ba5/greenlet-3.4.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:9b2d9a138ffa0e306d0e2b72976d2fb10b97e690d40ab36a472acaab0838e2de", size = 1573511, upload-time = "2026-04-08T16:26:20.058Z" },
     { url = "https://files.pythonhosted.org/packages/d4/0f/a5a26fe152fb3d12e6a474181f6e9848283504d0afd095f353d85726374b/greenlet-3.4.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:8424683caf46eb0eb6f626cb95e008e8cc30d0cb675bdfa48200925c79b38a08", size = 1640396, upload-time = "2026-04-08T15:57:30.88Z" },
     { url = "https://files.pythonhosted.org/packages/42/cf/bb2c32d9a100e36ee9f6e38fad6b1e082b8184010cb06259b49e1266ca01/greenlet-3.4.0-cp312-cp312-win_amd64.whl", hash = "sha256:a0a53fb071531d003b075c444014ff8f8b1a9898d36bb88abd9ac7b3524648a2", size = 238892, upload-time = "2026-04-08T17:03:10.094Z" },
@@ -513,6 +511,7 @@ dependencies = [
     { name = "hvac" },
     { name = "itsdangerous" },
     { name = "orjson" },
+    { name = "prometheus-client" },
     { name = "pydantic-settings" },
     { name = "python-jose" },
     { name = "python-ulid" },
@@ -535,6 +534,7 @@ requires-dist = [
     { name = "hvac", specifier = "==2.4.0" },
     { name = "itsdangerous", specifier = "==2.2.0" },
     { name = "orjson", specifier = "==3.11.7" },
+    { name = "prometheus-client", specifier = "==0.23.1" },
     { name = "pydantic-settings", specifier = "==2.12.0" },
     { name = "python-jose", specifier = "==3.5.0" },
     { name = "python-ulid", specifier = "==3.1.0" },
@@ -543,6 +543,15 @@ requires-dist = [
     { name = "uvloop", marker = "sys_platform != 'win32'", specifier = "==0.22.1" },
 ]
 
+[[package]]
+name = "prometheus-client"
+version = "0.23.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/23/53/3edb5d68ecf6b38fcbcc1ad28391117d2a322d9a1a3eff04bfdb184d8c3b/prometheus_client-0.23.1.tar.gz", hash = "sha256:6ae8f9081eaaaf153a2e959d2e6c4f4fb57b12ef76c8c7980202f1e57b48b2ce", size = 80481, upload-time = "2025-09-18T20:47:25.043Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b8/db/14bafcb4af2139e046d03fd00dea7873e48eafe18b7d2797e73d6681f210/prometheus_client-0.23.1-py3-none-any.whl", hash = "sha256:dd1913e6e76b59cfe44e7a4b83e01afc9873c1bdfd2ed8739f1e76aeca115f99", size = 61145, upload-time = "2025-09-18T20:47:23.875Z" },
+]
+
 [[package]]
 name = "propcache"
 version = "0.4.1"