diff --git a/Dockerfile b/Dockerfile index b05211b..76cf5dd 100644 --- a/Dockerfile +++ b/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"] diff --git a/pyproject.toml b/pyproject.toml index 7dbee6a..f8385ee 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/application/commands/payment_read_commands.py b/src/application/commands/payment_read_commands.py index 81da81d..a44c66e 100644 --- a/src/application/commands/payment_read_commands.py +++ b/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: diff --git a/src/infrastructure/database/repositories/order_repository.py b/src/infrastructure/database/repositories/order_repository.py index 8468861..c2a155a 100644 --- a/src/infrastructure/database/repositories/order_repository.py +++ b/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()] diff --git a/src/infrastructure/database/repositories/payment_repository.py b/src/infrastructure/database/repositories/payment_repository.py index 8647fcb..000f55a 100644 --- a/src/infrastructure/database/repositories/payment_repository.py +++ b/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()] diff --git a/src/infrastructure/database/repositories/sbp_withdrawal_repository.py b/src/infrastructure/database/repositories/sbp_withdrawal_repository.py index d9b6940..69e8206 100644 --- a/src/infrastructure/database/repositories/sbp_withdrawal_repository.py +++ b/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()] diff --git a/src/infrastructure/metrics.py b/src/infrastructure/metrics.py index e707a8f..27b1e52 100644 --- a/src/infrastructure/metrics.py +++ b/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) diff --git a/src/presentation/routing/order.py b/src/presentation/routing/order.py index aa283c0..e8b7e9c 100644 --- a/src/presentation/routing/order.py +++ b/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: diff --git a/src/presentation/schemas/order.py b/src/presentation/schemas/order.py index 94d17cf..a47c208 100644 --- a/src/presentation/schemas/order.py +++ b/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): diff --git a/uv.lock b/uv.lock index 6c74534..ff95b2d 100644 --- a/uv.lock +++ b/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"