diff --git a/docker-compose.yml b/docker-compose.yml index bdcb07b..15e9404 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -14,4 +14,44 @@ services: APP_WORKERS: "2" env_file: - .env - restart: no \ No newline at end of file + depends_on: + pay_keydb: + condition: service_healthy + restart: no + + pay_keydb: + image: eqalpha/keydb + container_name: pay_keydb + restart: no + expose: + - "6379" + volumes: + - pay_keydb_data:/data + command: + - keydb-server + - --requirepass + - ${REDIS_PASSWORD} + - --dir + - /data + - --appendonly + - "yes" + - --appendfsync + - everysec + - --save + - "900" + - "1" + - --save + - "300" + - "10" + - --save + - "60" + - "10000" + healthcheck: + test: ["CMD", "redis-cli", "-a", "${REDIS_PASSWORD}", "ping"] + interval: 5s + timeout: 2s + retries: 20 + + +volumes: + pay_keydb_data: \ No newline at end of file diff --git a/src/application/abstractions/i_unit_of_work.py b/src/application/abstractions/i_unit_of_work.py index 6cefe4a..2b542e5 100644 --- a/src/application/abstractions/i_unit_of_work.py +++ b/src/application/abstractions/i_unit_of_work.py @@ -1,6 +1,6 @@ from __future__ import annotations from typing import Protocol, runtime_checkable -from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,IUserRepository +from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,ISbpWithdrawalRepository,IUserRepository @runtime_checkable @@ -17,6 +17,9 @@ class IUnitOfWork(Protocol): @property def payment_repository(self) -> IPaymentRepository: ... + @property + def sbp_withdrawal_repository(self) -> ISbpWithdrawalRepository: ... + @property def user_repository(self) -> IUserRepository: ... diff --git a/src/application/abstractions/repositories/__init__.py b/src/application/abstractions/repositories/__init__.py index 7a85a0d..23cf501 100644 --- a/src/application/abstractions/repositories/__init__.py +++ b/src/application/abstractions/repositories/__init__.py @@ -1,3 +1,4 @@ from src.application.abstractions.repositories.i_order_repository import IOrderRepository from src.application.abstractions.repositories.i_payment_repository import IPaymentRepository +from src.application.abstractions.repositories.i_sbp_withdrawal_repository import ISbpWithdrawalRepository from src.application.abstractions.repositories.i_user_repository import IUserRepository \ No newline at end of file diff --git a/src/application/abstractions/repositories/i_sbp_withdrawal_repository.py b/src/application/abstractions/repositories/i_sbp_withdrawal_repository.py new file mode 100644 index 0000000..80e9210 --- /dev/null +++ b/src/application/abstractions/repositories/i_sbp_withdrawal_repository.py @@ -0,0 +1,47 @@ +from abc import ABC,abstractmethod +from src.application.domain.entities import SbpWithdrawalEntity +from src.application.domain.enums import SbpWithdrawalStatus + + +class ISbpWithdrawalRepository(ABC): + @abstractmethod + async def create(self,withdrawal: SbpWithdrawalEntity) -> SbpWithdrawalEntity: + raise NotImplementedError + + + @abstractmethod + async def get_by_id(self,withdraw_id: str) -> SbpWithdrawalEntity | None: + raise NotImplementedError + + + @abstractmethod + async def get_by_id_for_user(self,*,withdraw_id: str,user_id: str) -> SbpWithdrawalEntity | None: + raise NotImplementedError + + + @abstractmethod + async def list_by_user_id(self,*,user_id: str,limit: int,offset: int) -> list[SbpWithdrawalEntity]: + raise NotImplementedError + + + @abstractmethod + async def update_status( + self, + *, + withdraw_id: str, + status: SbpWithdrawalStatus, + provider_error: str | None = None, + ) -> None: + raise NotImplementedError + + + @abstractmethod + async def update_payout_created( + self, + *, + withdraw_id: str, + mozen_order_id: str | None, + status: SbpWithdrawalStatus, + provider_error: str | None, + ) -> None: + raise NotImplementedError diff --git a/src/application/commands/__init__.py b/src/application/commands/__init__.py index 1f6d0b7..45c951a 100644 --- a/src/application/commands/__init__.py +++ b/src/application/commands/__init__.py @@ -1,4 +1,5 @@ from src.application.commands.create_order_command import CreateOrderCommand from src.application.commands.create_payment_command import CreatePaymentCommand from src.application.commands.create_crypto_transfer_completed_command import CreateCryptoTransferCompletedCommand -from src.application.commands.payment_read_commands import GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,ListOrdersCommand,ListPaymentsCommand,OrderPaymentResult \ No newline at end of file +from src.application.commands.payment_read_commands import ClientOperationResult,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,ListClientOperationsCommand,ListOrdersCommand,ListPaymentsCommand,OrderPaymentResult +from src.application.commands.sbp_withdrawal_commands import CreateSbpWithdrawalCommand,GetSbpBanksCommand,GetSbpWithdrawalCommand,HandleSbpWithdrawalWalletEventCommand \ No newline at end of file diff --git a/src/application/commands/payment_read_commands.py b/src/application/commands/payment_read_commands.py index 47a274c..81da81d 100644 --- a/src/application/commands/payment_read_commands.py +++ b/src/application/commands/payment_read_commands.py @@ -1,9 +1,11 @@ from __future__ import annotations from dataclasses import dataclass +from datetime import datetime,timezone from decimal import Decimal +from typing import Literal from src.application.abstractions import IUnitOfWork from src.application.contracts import ILogger -from src.application.domain.entities import OrderEntity,PaymentEntity +from src.application.domain.entities import OrderEntity,PaymentEntity,SbpWithdrawalEntity from src.application.domain.exceptions import NotFoundException from src.application.services import PaymentQuote,PaymentQuoteService from src.infrastructure.database.decorators import transactional @@ -15,6 +17,14 @@ class OrderPaymentResult: payment: PaymentEntity | None +@dataclass(slots=True) +class ClientOperationResult: + type: Literal['payment','withdrawal'] + created_at: datetime + payment: PaymentEntity | None = None + withdrawal: SbpWithdrawalEntity | None = None + + class GetPaymentConfigCommand: def __init__(self, *, payment_quote_service: PaymentQuoteService, logger: ILogger): self._payment_quote_service = payment_quote_service @@ -91,6 +101,39 @@ class ListPaymentsCommand: return payments +class ListClientOperationsCommand: + def __init__(self, *, unit_of_work: IUnitOfWork, logger: ILogger): + self._unit_of_work = unit_of_work + self._logger = logger + + + @transactional + async def __call__(self, *, user_id: str, limit: int, offset: int) -> list[ClientOperationResult]: + fetch_limit=limit+offset + payments=await self._unit_of_work.payment_repository.list_by_user_id( + user_id=user_id, + limit=fetch_limit, + offset=0, + ) + withdrawals=await self._unit_of_work.sbp_withdrawal_repository.list_by_user_id( + user_id=user_id, + limit=fetch_limit, + offset=0, + ) + min_dt=datetime.min.replace(tzinfo=timezone.utc) + items=[ + ClientOperationResult(type='payment',created_at=payment.created_at or min_dt,payment=payment) + for payment in payments + ] + items.extend( + ClientOperationResult(type='withdrawal',created_at=withdrawal.created_at or min_dt,withdrawal=withdrawal) + 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] + + class GetPaymentCommand: def __init__(self, *, unit_of_work: IUnitOfWork, logger: ILogger): self._unit_of_work = unit_of_work diff --git a/src/application/commands/sbp_withdrawal_commands.py b/src/application/commands/sbp_withdrawal_commands.py new file mode 100644 index 0000000..c932d09 --- /dev/null +++ b/src/application/commands/sbp_withdrawal_commands.py @@ -0,0 +1,254 @@ +from __future__ import annotations +from datetime import datetime,timezone +from decimal import Decimal +from typing import Any +from ulid import ULID +from src.application.abstractions import IUnitOfWork +from src.application.contracts import IQueueMessanger,ILogger,IMozenSbpService,SbpBank +from src.application.domain.entities import SbpWithdrawalEntity +from src.application.domain.enums import SbpWithdrawalStatus +from src.application.domain.exceptions import BadRequestException,NotFoundException,PaymentProviderException,ServiceUnavailableException +from src.application.services import SbpWithdrawalQuoteService +from src.infrastructure.config import settings +from src.infrastructure.database.decorators import transactional + + +class GetSbpBanksCommand: + def __init__(self,*,mozen_sbp_service: IMozenSbpService,logger: ILogger): + self._mozen_sbp_service=mozen_sbp_service + self._logger=logger + + + async def __call__(self) -> list[SbpBank]: + return await self._mozen_sbp_service.get_banks(self._logger.get_trace_id()) + + +class CreateSbpWithdrawalCommand: + def __init__( + self, + *, + unit_of_work: IUnitOfWork, + logger: ILogger, + queue_messanger: IQueueMessanger, + mozen_sbp_service: IMozenSbpService, + quote_service: SbpWithdrawalQuoteService, + ): + self._unit_of_work=unit_of_work + self._logger=logger + self._queue_messanger=queue_messanger + self._mozen_sbp_service=mozen_sbp_service + self._quote_service=quote_service + + + async def _bank_name(self,bank_id: str) -> str: + banks=await self._mozen_sbp_service.get_banks(self._logger.get_trace_id()) + for bank in banks: + if bank.id==bank_id: + return bank.name + raise BadRequestException(message='SBP bank not found') + + + @transactional + async def __call__(self,*,user_id: str,bank_id: str,phone: str,usdt_amount: Decimal) -> SbpWithdrawalEntity: + user=await self._unit_of_work.user_repository.get(user_id) + if user is None: + raise NotFoundException(message='User not found') + wallet_address=str(settings.SBP_WITHDRAWAL_WALLET_ADDRESS or '').strip() + if not wallet_address: + raise ServiceUnavailableException(message='SBP withdrawal wallet address unavailable') + trace_id=self._logger.get_trace_id() + withdraw_id=str(ULID()) + bank_name=await self._bank_name(bank_id) + quote=await self._quote_service.get_quote(usdt_amount) + withdrawal=SbpWithdrawalEntity( + id=withdraw_id, + user_id=user_id, + bank_id=bank_id, + bank_name=bank_name, + phone=phone, + wallet_address=wallet_address, + usdt_amount=quote.usdt_amount, + rub_amount=quote.rub_amount, + usdt_exchange_rate=quote.usdt_exchange_rate, + service_fee_rate=quote.service_fee_rate, + service_fee_usdt=quote.service_fee_usdt, + trace_id=trace_id, + status=SbpWithdrawalStatus.WAITING_USDT, + created_at=datetime.now(timezone.utc), + ) + saved=await self._unit_of_work.sbp_withdrawal_repository.create(withdrawal) + message_id=str(ULID()) + message: dict[str,Any]={ + 'user_id':user_id, + 'usdt_amount':str(quote.usdt_amount), + 'trace_id':trace_id, + 'wallet_address':wallet_address, + 'withdraw_id':withdraw_id, + 'message_id':message_id, + } + await self._queue_messanger.publish_to_queue( + queue=settings.RABBIT_SBP_WITHDRAWAL_USDT_REQUESTED_QUEUE, + message=message, + message_id=message_id, + correlation_id=trace_id, + headers={'trace_id':trace_id}, + ) + self._logger.info({ + 'event':'sbp_withdrawal_usdt_request_published', + 'withdraw_id':withdraw_id, + 'user_id':user_id, + 'message_id':message_id, + }) + return saved + + +class GetSbpWithdrawalCommand: + def __init__(self,*,unit_of_work: IUnitOfWork,logger: ILogger): + self._unit_of_work=unit_of_work + self._logger=logger + + + @transactional + async def __call__(self,*,withdraw_id: str,user_id: str) -> SbpWithdrawalEntity: + withdrawal=await self._unit_of_work.sbp_withdrawal_repository.get_by_id_for_user( + withdraw_id=withdraw_id, + user_id=user_id, + ) + if withdrawal is None: + raise NotFoundException(message='SBP withdrawal not found') + return withdrawal + + +class HandleSbpWithdrawalWalletEventCommand: + def __init__( + self, + *, + unit_of_work: IUnitOfWork, + logger: ILogger, + mozen_sbp_service: IMozenSbpService, + ): + self._unit_of_work=unit_of_work + self._logger=logger + self._mozen_sbp_service=mozen_sbp_service + + + @staticmethod + def _event_error( + *, + event_type: str, + received_usdt_amount: Decimal | None, + tx_hash: str | None, + error: str | None, + ) -> str | None: + parts=( + f'event_type={event_type}', + f'received_usdt_amount={received_usdt_amount}' if received_usdt_amount is not None else '', + f'tx_hash={tx_hash}' if tx_hash else '', + f'error={error}' if error else '', + ) + value=';'.join(part for part in parts if part) + return value or None + + + @transactional + async def __call__( + self, + *, + withdraw_id: str, + user_id: str, + event_type: str, + received_usdt_amount: Decimal | None = None, + tx_hash: str | None = None, + error: str | None = None, + ) -> None: + withdrawal=await self._unit_of_work.sbp_withdrawal_repository.get_by_id(withdraw_id) + if withdrawal is None: + raise NotFoundException(message='SBP withdrawal not found') + if withdrawal.user_id!=user_id: + raise BadRequestException(message='SBP withdrawal user mismatch') + normalized_event_type=event_type.strip().lower() + if withdrawal.status in ( + SbpWithdrawalStatus.PAYOUT_PROCESSING, + SbpWithdrawalStatus.PAYOUT_COMPLETED, + SbpWithdrawalStatus.PAYOUT_FAILED, + SbpWithdrawalStatus.CANCELLED, + SbpWithdrawalStatus.EXPIRED, + ): + self._logger.info({ + 'event':'sbp_withdrawal_wallet_event_ignored', + 'withdraw_id':withdraw_id, + 'user_id':user_id, + 'wallet_event_type':normalized_event_type, + 'status':withdrawal.status.value if withdrawal.status is not None else None, + }) + return + provider_error=self._event_error( + event_type=normalized_event_type, + received_usdt_amount=received_usdt_amount, + tx_hash=tx_hash, + error=error, + ) + if normalized_event_type in ('usdt_underpaid','underpaid'): + await self._unit_of_work.sbp_withdrawal_repository.update_status( + withdraw_id=withdraw_id, + status=SbpWithdrawalStatus.USDT_UNDERPAID, + provider_error=provider_error, + ) + return + if normalized_event_type in ('expired','usdt_expired'): + await self._unit_of_work.sbp_withdrawal_repository.update_status( + withdraw_id=withdraw_id, + status=SbpWithdrawalStatus.EXPIRED, + provider_error=provider_error, + ) + return + if normalized_event_type in ('failed','wallet_error','watch_failed','error'): + await self._unit_of_work.sbp_withdrawal_repository.update_status( + withdraw_id=withdraw_id, + status=SbpWithdrawalStatus.WALLET_ERROR, + provider_error=provider_error, + ) + return + if normalized_event_type not in ('usdt_received','received','confirmed'): + raise BadRequestException(message='Unknown SBP withdrawal wallet event type') + await self._unit_of_work.sbp_withdrawal_repository.update_status( + withdraw_id=withdraw_id, + status=SbpWithdrawalStatus.USDT_RECEIVED, + provider_error=provider_error, + ) + try: + result=await self._mozen_sbp_service.make_withdrawal( + user_id=user_id, + withdraw_id=withdraw_id, + phone=str(withdrawal.phone or ''), + bank_id=str(withdrawal.bank_id or ''), + # amount=Decimal(str(withdrawal.rub_amount or '0')), + amount=Decimal('1'), + trace_id=str(withdrawal.trace_id or self._logger.get_trace_id()), + ) + except PaymentProviderException as exception: + await self._unit_of_work.sbp_withdrawal_repository.update_status( + withdraw_id=withdraw_id, + status=SbpWithdrawalStatus.PAYOUT_FAILED, + provider_error=exception.message, + ) + return + status=SbpWithdrawalStatus.PAYOUT_PROCESSING + provider_error=result.error + if result.error: + status=SbpWithdrawalStatus.PAYOUT_FAILED + elif result.status.strip().lower() in ('completed','success','succeeded','paid'): + status=SbpWithdrawalStatus.PAYOUT_COMPLETED + await self._unit_of_work.sbp_withdrawal_repository.update_payout_created( + withdraw_id=withdraw_id, + mozen_order_id=result.mozen_order_id, + status=status, + provider_error=provider_error, + ) + self._logger.info({ + 'event':'sbp_withdrawal_payout_created', + 'withdraw_id':withdraw_id, + 'user_id':user_id, + 'mozen_order_id':result.mozen_order_id, + 'status':status.value, + }) diff --git a/src/application/contracts/__init__.py b/src/application/contracts/__init__.py index 8f9d6a6..5334c0d 100644 --- a/src/application/contracts/__init__.py +++ b/src/application/contracts/__init__.py @@ -5,4 +5,5 @@ from src.application.contracts.i_cache import ICache from src.application.contracts.i_hash_service import IHashService from src.application.contracts.i_queue_messanger import IQueueMessanger from src.application.contracts.i_itpay_service import IItPayService -from src.application.contracts.i_receipt import IReceipt \ No newline at end of file +from src.application.contracts.i_receipt import IReceipt +from src.application.contracts.i_mozen_sbp_service import IMozenSbpService,SbpBank,SbpPayoutResult \ No newline at end of file diff --git a/src/application/contracts/i_mozen_sbp_service.py b/src/application/contracts/i_mozen_sbp_service.py new file mode 100644 index 0000000..13321a3 --- /dev/null +++ b/src/application/contracts/i_mozen_sbp_service.py @@ -0,0 +1,36 @@ +from abc import ABC,abstractmethod +from dataclasses import dataclass +from decimal import Decimal + + +@dataclass(slots=True) +class SbpBank: + id: str + name: str + + +@dataclass(slots=True) +class SbpPayoutResult: + status: str + mozen_order_id: str | None = None + error: str | None = None + + +class IMozenSbpService(ABC): + @abstractmethod + async def get_banks(self,trace_id: str) -> list[SbpBank]: + raise NotImplementedError + + + @abstractmethod + async def make_withdrawal( + self, + *, + user_id: str, + withdraw_id: str, + phone: str, + bank_id: str, + amount: Decimal, + trace_id: str, + ) -> SbpPayoutResult: + raise NotImplementedError diff --git a/src/application/domain/entities/__init__.py b/src/application/domain/entities/__init__.py index 88fc910..eba7c5e 100644 --- a/src/application/domain/entities/__init__.py +++ b/src/application/domain/entities/__init__.py @@ -1,5 +1,6 @@ from src.application.domain.entities.order import OrderEntity from src.application.domain.entities.payment import PaymentEntity +from src.application.domain.entities.sbp_withdrawal import SbpWithdrawalEntity -__all__ = ['PaymentEntity', 'OrderEntity'] \ No newline at end of file +__all__ = ['PaymentEntity','OrderEntity','SbpWithdrawalEntity'] \ No newline at end of file diff --git a/src/application/domain/entities/sbp_withdrawal.py b/src/application/domain/entities/sbp_withdrawal.py new file mode 100644 index 0000000..c8d64b6 --- /dev/null +++ b/src/application/domain/entities/sbp_withdrawal.py @@ -0,0 +1,30 @@ +from __future__ import annotations +from dataclasses import dataclass +from datetime import datetime +from decimal import Decimal +from src.application.domain.enums import SbpWithdrawalStatus + + +@dataclass(slots=True) +class SbpWithdrawalEntity: + id: str | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + + user_id: str | None = None + bank_id: str | None = None + bank_name: str | None = None + phone: str | None = None + wallet_address: str | None = None + usdt_amount: Decimal | None = None + rub_amount: Decimal | None = None + usdt_exchange_rate: Decimal | None = None + service_fee_rate: Decimal | None = None + service_fee_usdt: Decimal | None = None + mozen_order_id: str | None = None + trace_id: str | None = None + status: SbpWithdrawalStatus | None = None + provider_error: str | None = None + usdt_received_at: datetime | None = None + payout_created_at: datetime | None = None + payout_completed_at: datetime | None = None diff --git a/src/application/domain/enums/__init__.py b/src/application/domain/enums/__init__.py index 0dbe6df..023cf3f 100644 --- a/src/application/domain/enums/__init__.py +++ b/src/application/domain/enums/__init__.py @@ -2,4 +2,5 @@ from src.application.domain.enums.log_level import LogLevel from src.application.domain.enums.log_format import LogFormat from src.application.domain.enums.itpay_payment_status import ItPayPaymentStatus from src.application.domain.enums.order_status import OrderStatus -from src.application.domain.enums.payment_status import PaymentStatus \ No newline at end of file +from src.application.domain.enums.payment_status import PaymentStatus +from src.application.domain.enums.sbp_withdrawal_status import SbpWithdrawalStatus \ No newline at end of file diff --git a/src/application/domain/enums/sbp_withdrawal_status.py b/src/application/domain/enums/sbp_withdrawal_status.py new file mode 100644 index 0000000..40286f9 --- /dev/null +++ b/src/application/domain/enums/sbp_withdrawal_status.py @@ -0,0 +1,14 @@ +from enum import Enum + + +class SbpWithdrawalStatus(str,Enum): + CREATED='created' + WAITING_USDT='waiting_usdt' + USDT_UNDERPAID='usdt_underpaid' + USDT_RECEIVED='usdt_received' + WALLET_ERROR='wallet_error' + PAYOUT_PROCESSING='payout_processing' + PAYOUT_COMPLETED='payout_completed' + PAYOUT_FAILED='payout_failed' + CANCELLED='cancelled' + EXPIRED='expired' diff --git a/src/application/services/__init__.py b/src/application/services/__init__.py index 99e00c0..aba536e 100644 --- a/src/application/services/__init__.py +++ b/src/application/services/__init__.py @@ -1,4 +1,5 @@ from src.application.services.payment_quote_service import PaymentQuote,PaymentQuoteService +from src.application.services.sbp_withdrawal_quote_service import SbpWithdrawalQuote,SbpWithdrawalQuoteService __all__ = ['PaymentQuote', 'PaymentQuoteService'] diff --git a/src/application/services/sbp_withdrawal_quote_service.py b/src/application/services/sbp_withdrawal_quote_service.py new file mode 100644 index 0000000..0a9d645 --- /dev/null +++ b/src/application/services/sbp_withdrawal_quote_service.py @@ -0,0 +1,64 @@ +from __future__ import annotations +from dataclasses import dataclass +from decimal import Decimal,ROUND_DOWN,ROUND_UP +from src.application.contracts import ICache,ILogger +from src.application.domain.exceptions import OrderTotalOutOfRangeException,ServiceUnavailableException + +MIN_SBP_WITHDRAWAL_USDT: Decimal=Decimal('100') +MAX_SBP_WITHDRAWAL_USDT: Decimal=Decimal('5000') + +_FEE_TIERS: tuple[tuple[Decimal,Decimal,Decimal],...]=( + (Decimal('100'),Decimal('2000'),Decimal('0.08')), + (Decimal('2000.01'),Decimal('4000'),Decimal('0.06')), + (Decimal('4000.01'),Decimal('5000'),Decimal('0.04')), +) + + +@dataclass(slots=True) +class SbpWithdrawalQuote: + usdt_amount: Decimal + rub_amount: Decimal + usdt_exchange_rate: Decimal + service_fee_rate: Decimal + service_fee_usdt: Decimal + + +class SbpWithdrawalQuoteService: + def __init__(self,*,remote_cache: ICache,logger: ILogger): + self._remote_cache=remote_cache + self._logger=logger + + + async def _load_rate(self) -> Decimal: + rate_raw=await self._remote_cache.hget('tradex:rub:rate','value') + self._logger.info(f'Actual market value for SBP withdrawal: rate={rate_raw}') + if rate_raw is None: + raise ServiceUnavailableException(message='Exchange rate unavailable') + return Decimal(rate_raw).quantize(Decimal('0.00'),rounding=ROUND_UP) + + + @staticmethod + def _fee_rate(usdt_amount: Decimal) -> Decimal: + for lo,hi,rate in _FEE_TIERS: + if lo<=usdt_amount<=hi: + return rate + raise OrderTotalOutOfRangeException(message='Withdrawal amount is out of range') + + + async def get_quote(self,usdt_amount: Decimal) -> SbpWithdrawalQuote: + if usdt_amountMAX_SBP_WITHDRAWAL_USDT: + raise OrderTotalOutOfRangeException(message='Withdrawal amount exceeds maximum allowed amount') + rate=await self._load_rate() + fee_rate=self._fee_rate(usdt_amount) + service_fee=(usdt_amount*fee_rate).quantize(Decimal('0.01')) + payout_usdt=(usdt_amount-service_fee).quantize(Decimal('0.01')) + rub_amount=(payout_usdt*rate).quantize(Decimal('1'),rounding=ROUND_DOWN) + return SbpWithdrawalQuote( + usdt_amount=usdt_amount, + rub_amount=rub_amount, + usdt_exchange_rate=rate, + service_fee_rate=fee_rate, + service_fee_usdt=service_fee, + ) diff --git a/src/infrastructure/config/settings.py b/src/infrastructure/config/settings.py index 92f0316..062d24d 100644 --- a/src/infrastructure/config/settings.py +++ b/src/infrastructure/config/settings.py @@ -90,10 +90,21 @@ class Settings(BaseSettings): RABBIT_EMAIL_CODE_QUEUE: str = "email.verification_code" RABBIT_CRYPTO_TRANSFER_QUEUE: str = "crypto.transfer.requested" RABBIT_CRYPTO_TRANSFER_COMPLETED_QUEUE: str = "crypto.transfer.completed" + RABBIT_SBP_WITHDRAWAL_USDT_REQUESTED_QUEUE: str = 'sbp.withdrawal.usdt.requested' + RABBIT_SBP_WITHDRAWAL_WALLET_EVENTS_QUEUE: str = 'sbp.withdrawal.wallet.events' ITPAY_PUBLIC_ID: str ITPAY_API_SECRET: str + MOZEN_API_BASE_URL: str = 'https://extgw.mozen.io' + MOZEN_API_SECRET: str = '' + MOZEN_CLIENT_ID: str = '' + MOZEN_ENDPOINT_ID: str = '' + MOZEN_MERCHANT_ID: str = '' + MOZEN_PARK_ID: str = '' + MOZEN_PARK_TOKEN: str = '' + SBP_WITHDRAWAL_WALLET_ADDRESS: str = '' + CLOUD_KASSIR_PUBLIC_ID: str = '' CLOUD_KASSIR_API_SECRET: str = '' @@ -210,6 +221,28 @@ class Settings(BaseSettings): rb_set('password', 'RABBIT_PASSWORD') rb_set('vhost', 'RABBIT_VHOST') + extgw = read_secret_optional('extgw') + if extgw: + extgw_ci = {str(k).lower(): v for k, v in extgw.items()} + + def extgw_set(field: str, env_key: str) -> None: + v = extgw_ci.get(field) + if v is None: + return + if isinstance(v, str) and not v.strip(): + return + data[env_key] = str(v).strip() + + extgw_set('api_base_url', 'MOZEN_API_BASE_URL') + extgw_set('api_secret', 'MOZEN_API_SECRET') + extgw_set('client_id', 'MOZEN_CLIENT_ID') + extgw_set('endpoint_id', 'MOZEN_ENDPOINT_ID') + extgw_set('merchant_id', 'MOZEN_MERCHANT_ID') + extgw_set('park_id', 'MOZEN_PARK_ID') + extgw_set('park_token', 'MOZEN_PARK_TOKEN') + extgw_set('wallet_address', 'SBP_WITHDRAWAL_WALLET_ADDRESS') + extgw_set('usdt_wallet_address', 'SBP_WITHDRAWAL_WALLET_ADDRESS') + keydb = read_secret('keydb') k_ci = {str(k).lower(): v for k, v in keydb.items()} diff --git a/src/infrastructure/database/models/__init__.py b/src/infrastructure/database/models/__init__.py index 16813d6..b64fc01 100644 --- a/src/infrastructure/database/models/__init__.py +++ b/src/infrastructure/database/models/__init__.py @@ -1,7 +1,8 @@ from src.infrastructure.database.models.base import Base from src.infrastructure.database.models.order import Order from src.infrastructure.database.models.payment import Payment +from src.infrastructure.database.models.sbp_withdrawal import SbpWithdrawal from src.infrastructure.database.models.user import UserModel -__all__ = ['Base','Order','Payment','UserModel'] +__all__ = ['Base','Order','Payment','SbpWithdrawal','UserModel'] diff --git a/src/infrastructure/database/models/sbp_withdrawal.py b/src/infrastructure/database/models/sbp_withdrawal.py new file mode 100644 index 0000000..a41c890 --- /dev/null +++ b/src/infrastructure/database/models/sbp_withdrawal.py @@ -0,0 +1,40 @@ +from __future__ import annotations +from datetime import datetime +from decimal import Decimal +from sqlalchemy import DateTime,Enum as SAEnum,ForeignKey,Numeric,String,Text +from sqlalchemy.orm import Mapped,mapped_column +from src.application.domain.enums import SbpWithdrawalStatus +from src.infrastructure.database.models.base import Base +from src.infrastructure.database.models.mixins import AuditTimestampsMixin,UlidPrimaryKeyMixin + + +class SbpWithdrawal(Base,UlidPrimaryKeyMixin,AuditTimestampsMixin): + __tablename__='sbp_withdrawals' + + user_id: Mapped[str]=mapped_column( + String(26), + ForeignKey('users.id',ondelete='RESTRICT'), + nullable=False, + index=True, + ) + bank_id: Mapped[str]=mapped_column(String(32),nullable=False,index=True) + bank_name: Mapped[str]=mapped_column(String(255),nullable=False) + phone: Mapped[str]=mapped_column(String(32),nullable=False) + wallet_address: Mapped[str]=mapped_column(String(255),nullable=False) + usdt_amount: Mapped[Decimal]=mapped_column(Numeric(38,2),nullable=False) + rub_amount: Mapped[Decimal]=mapped_column(Numeric(38,2),nullable=False) + usdt_exchange_rate: Mapped[Decimal]=mapped_column(Numeric(38,2),nullable=False) + service_fee_rate: Mapped[Decimal]=mapped_column(Numeric(8,4),nullable=False) + service_fee_usdt: Mapped[Decimal]=mapped_column(Numeric(38,2),nullable=False) + mozen_order_id: Mapped[str | None]=mapped_column(String(128),nullable=True,index=True) + trace_id: Mapped[str]=mapped_column(String(64),nullable=False,index=True) + status: Mapped[SbpWithdrawalStatus]=mapped_column( + SAEnum(SbpWithdrawalStatus,name='sbp_withdrawal_status_enum',values_callable=lambda x:[e.value for e in x]), + nullable=False, + index=True, + default=SbpWithdrawalStatus.CREATED, + ) + provider_error: Mapped[str | None]=mapped_column(Text,nullable=True) + usdt_received_at: Mapped[datetime | None]=mapped_column(DateTime(timezone=True),nullable=True) + payout_created_at: Mapped[datetime | None]=mapped_column(DateTime(timezone=True),nullable=True) + payout_completed_at: Mapped[datetime | None]=mapped_column(DateTime(timezone=True),nullable=True) diff --git a/src/infrastructure/database/repositories/sbp_withdrawal_repository.py b/src/infrastructure/database/repositories/sbp_withdrawal_repository.py new file mode 100644 index 0000000..f357f9e --- /dev/null +++ b/src/infrastructure/database/repositories/sbp_withdrawal_repository.py @@ -0,0 +1,139 @@ +from __future__ import annotations +from dataclasses import replace +from datetime import datetime,timezone +from sqlalchemy import desc,select +from sqlalchemy.ext.asyncio import AsyncSession +from src.application.abstractions.repositories import ISbpWithdrawalRepository +from src.application.contracts import ILogger +from src.application.domain.entities import SbpWithdrawalEntity +from src.application.domain.enums import SbpWithdrawalStatus +from src.infrastructure.database.models.sbp_withdrawal import SbpWithdrawal + + +class SbpWithdrawalRepository(ISbpWithdrawalRepository): + def __init__(self,session: AsyncSession,logger: ILogger): + self._session=session + self._logger=logger + + + @staticmethod + def _to_entity(model: SbpWithdrawal) -> SbpWithdrawalEntity: + return SbpWithdrawalEntity( + id=model.id, + created_at=model.created_at, + updated_at=model.updated_at, + user_id=model.user_id, + bank_id=model.bank_id, + bank_name=model.bank_name, + phone=model.phone, + wallet_address=model.wallet_address, + usdt_amount=model.usdt_amount, + rub_amount=model.rub_amount, + usdt_exchange_rate=model.usdt_exchange_rate, + service_fee_rate=model.service_fee_rate, + service_fee_usdt=model.service_fee_usdt, + mozen_order_id=model.mozen_order_id, + trace_id=model.trace_id, + status=model.status, + provider_error=model.provider_error, + usdt_received_at=model.usdt_received_at, + payout_created_at=model.payout_created_at, + payout_completed_at=model.payout_completed_at, + ) + + + async def create(self,withdrawal: SbpWithdrawalEntity) -> SbpWithdrawalEntity: + values={ + 'user_id':withdrawal.user_id, + 'bank_id':withdrawal.bank_id, + 'bank_name':withdrawal.bank_name, + 'phone':withdrawal.phone, + 'wallet_address':withdrawal.wallet_address, + 'usdt_amount':withdrawal.usdt_amount, + 'rub_amount':withdrawal.rub_amount, + 'usdt_exchange_rate':withdrawal.usdt_exchange_rate, + 'service_fee_rate':withdrawal.service_fee_rate, + 'service_fee_usdt':withdrawal.service_fee_usdt, + 'mozen_order_id':withdrawal.mozen_order_id, + 'trace_id':withdrawal.trace_id, + 'status':withdrawal.status or SbpWithdrawalStatus.CREATED, + 'provider_error':withdrawal.provider_error, + 'usdt_received_at':withdrawal.usdt_received_at, + 'payout_created_at':withdrawal.payout_created_at, + 'payout_completed_at':withdrawal.payout_completed_at, + } + if withdrawal.id is not None: + values['id']=withdrawal.id + model=SbpWithdrawal(**values) + self._session.add(model) + await self._session.flush() + return replace(withdrawal,id=model.id,created_at=model.created_at,updated_at=model.updated_at) + + + async def get_by_id(self,withdraw_id: str) -> SbpWithdrawalEntity | None: + stmt=select(SbpWithdrawal).where(SbpWithdrawal.id==withdraw_id) + model=await self._session.scalar(stmt) + if model is None: + return None + return self._to_entity(model) + + + async def get_by_id_for_user(self,*,withdraw_id: str,user_id: str) -> SbpWithdrawalEntity | None: + stmt=select(SbpWithdrawal).where(SbpWithdrawal.id==withdraw_id,SbpWithdrawal.user_id==user_id) + model=await self._session.scalar(stmt) + if model is None: + return None + return self._to_entity(model) + + + async def list_by_user_id(self,*,user_id: str,limit: int,offset: int) -> list[SbpWithdrawalEntity]: + stmt=( + select(SbpWithdrawal) + .where(SbpWithdrawal.user_id==user_id) + .order_by(desc(SbpWithdrawal.created_at)) + .limit(limit) + .offset(offset) + ) + result=await self._session.scalars(stmt) + return [self._to_entity(model) for model in result.all()] + + + async def update_status( + self, + *, + withdraw_id: str, + status: SbpWithdrawalStatus, + provider_error: str | None = None, + ) -> None: + model=await self._session.get(SbpWithdrawal,withdraw_id) + if model is None: + return + model.status=status + model.provider_error=provider_error + if status==SbpWithdrawalStatus.USDT_RECEIVED: + model.usdt_received_at=datetime.now(timezone.utc) + if status==SbpWithdrawalStatus.PAYOUT_PROCESSING: + model.payout_created_at=datetime.now(timezone.utc) + if status==SbpWithdrawalStatus.PAYOUT_COMPLETED: + model.payout_completed_at=datetime.now(timezone.utc) + await self._session.flush() + + + async def update_payout_created( + self, + *, + withdraw_id: str, + mozen_order_id: str | None, + status: SbpWithdrawalStatus, + provider_error: str | None, + ) -> None: + model=await self._session.get(SbpWithdrawal,withdraw_id) + if model is None: + return + model.mozen_order_id=mozen_order_id + model.status=status + model.provider_error=provider_error + model.payout_created_at=datetime.now(timezone.utc) + if status==SbpWithdrawalStatus.PAYOUT_COMPLETED: + model.payout_completed_at=datetime.now(timezone.utc) + await self._session.flush() diff --git a/src/infrastructure/database/unit_of_work.py b/src/infrastructure/database/unit_of_work.py index a416698..1db7df9 100644 --- a/src/infrastructure/database/unit_of_work.py +++ b/src/infrastructure/database/unit_of_work.py @@ -1,9 +1,10 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.application.abstractions import IUnitOfWork -from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,IUserRepository +from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,ISbpWithdrawalRepository,IUserRepository from src.application.contracts import ILogger from src.infrastructure.database.repositories.order_repository import OrderRepository from src.infrastructure.database.repositories.payment_repository import PaymentRepository +from src.infrastructure.database.repositories.sbp_withdrawal_repository import SbpWithdrawalRepository from src.infrastructure.database.repositories.user_repository import UserRepository @@ -14,6 +15,7 @@ class UnitOfWork(IUnitOfWork): self._session: AsyncSession = None self._order_repository: IOrderRepository | None = None self._payment_repository: IPaymentRepository | None = None + self._sbp_withdrawal_repository: ISbpWithdrawalRepository | None = None self._user_repository: IUserRepository | None = None self._logger: ILogger = logger @@ -21,6 +23,7 @@ class UnitOfWork(IUnitOfWork): self._session = self.session_factory() self._order_repository = None self._payment_repository = None + self._sbp_withdrawal_repository = None self._user_repository = None return self @@ -49,6 +52,13 @@ class UnitOfWork(IUnitOfWork): return self._payment_repository + @property + def sbp_withdrawal_repository(self) -> ISbpWithdrawalRepository: + if self._sbp_withdrawal_repository is None: + self._sbp_withdrawal_repository = SbpWithdrawalRepository(session=self._session,logger=self._logger) + return self._sbp_withdrawal_repository + + @property def user_repository(self) -> IUserRepository: if self._user_repository is None: diff --git a/src/infrastructure/mozen/__init__.py b/src/infrastructure/mozen/__init__.py new file mode 100644 index 0000000..87e2ce7 --- /dev/null +++ b/src/infrastructure/mozen/__init__.py @@ -0,0 +1,4 @@ +from src.infrastructure.mozen.sbp_client import MozenSbpClient + + +__all__=['MozenSbpClient'] diff --git a/src/infrastructure/mozen/sbp_client.py b/src/infrastructure/mozen/sbp_client.py new file mode 100644 index 0000000..736ba60 --- /dev/null +++ b/src/infrastructure/mozen/sbp_client.py @@ -0,0 +1,115 @@ +from __future__ import annotations +from decimal import Decimal,ROUND_DOWN +from typing import Any +import aiohttp +import orjson +from aiohttp import ClientTimeout +from src.application.contracts import IMozenSbpService,SbpBank,SbpPayoutResult +from src.application.domain.exceptions import PaymentProviderException,ServiceUnavailableException + + +class MozenSbpClient(IMozenSbpService): + def __init__( + self, + *, + api_base_url: str, + client_id: str, + park_token: str, + timeout_seconds: float = 30, + ) -> None: + self._api_base_url=api_base_url.rstrip('/') + self._client_id=client_id + self._park_token=park_token + self._timeout=ClientTimeout(total=timeout_seconds) + + + def _auth_headers(self,trace_id: str) -> dict[str,str]: + if not self._client_id or not self._park_token: + raise ServiceUnavailableException(message='Mozen SBP credentials unavailable') + return { + 'Accept':'application/json', + 'Content-Type':'application/json', + 'x-trace-id':trace_id, + 'x-client-id':self._client_id, + 'park-token':self._park_token, + } + + + async def get_banks(self,trace_id: str) -> list[SbpBank]: + url=f'{self._api_base_url}/api/v2/payout/sbp/banks' + headers={'Accept':'application/json','x-trace-id':trace_id} + try: + async with aiohttp.ClientSession(timeout=self._timeout) as session: + async with session.get(url,headers=headers) as resp: + response_text=await resp.text() + try: + response_json=orjson.loads(response_text) + except orjson.JSONDecodeError: + response_json=[] + if resp.status>=400: + raise PaymentProviderException(message='Mozen SBP banks provider error') + if not isinstance(response_json,list): + raise PaymentProviderException(message='Mozen SBP banks response invalid') + banks: list[SbpBank]=[] + for item in response_json: + if not isinstance(item,dict): + continue + bank_id=str(item.get('id') or '').strip() + name=str(item.get('name') or '').strip() + if bank_id and name: + banks.append(SbpBank(id=bank_id,name=name)) + return banks + except PaymentProviderException: + raise + except aiohttp.ClientError: + raise PaymentProviderException(message='Mozen SBP banks provider unreachable') + + + async def make_withdrawal( + self, + *, + user_id: str, + withdraw_id: str, + phone: str, + bank_id: str, + amount: Decimal, + trace_id: str, + ) -> SbpPayoutResult: + amount_int=int(amount.quantize(Decimal('1'),rounding=ROUND_DOWN)) + payload: dict[str,Any]={ + 'amount':amount_int, + 'order_id':withdraw_id, + 'phone':phone, + 'bank_id':bank_id, + 'description':'SBP withdrawal', + 'user_id':user_id, + } + url=f'{self._api_base_url}/api/v2/payout/sbp/withdrawal' + try: + async with aiohttp.ClientSession(timeout=self._timeout) as session: + async with session.post(url,json=payload,headers=self._auth_headers(trace_id)) as resp: + response_text=await resp.text() + try: + response_json=orjson.loads(response_text) + except orjson.JSONDecodeError: + response_json={'raw':response_text} + if resp.status>=400: + raise PaymentProviderException(message='Mozen SBP withdrawal provider error') + if not isinstance(response_json,dict): + raise PaymentProviderException(message='Mozen SBP withdrawal response invalid') + error=response_json.get('error') + if isinstance(error,dict): + error_message=str(error.get('message') or error.get('code') or '').strip() or 'Mozen SBP withdrawal rejected' + return SbpPayoutResult(status=str(response_json.get('status') or 'error'),error=error_message) + request=response_json.get('withdrawal_request') + mozen_order_id=None + if isinstance(request,dict): + mozen_order_id=str(request.get('id') or '').strip() or None + return SbpPayoutResult( + status=str(response_json.get('status') or '').strip(), + mozen_order_id=mozen_order_id, + ) + except PaymentProviderException: + raise + except aiohttp.ClientError: + raise PaymentProviderException(message='Mozen SBP withdrawal provider unreachable') diff --git a/src/main.py b/src/main.py index 08f627a..8e22621 100644 --- a/src/main.py +++ b/src/main.py @@ -15,9 +15,9 @@ from src.infrastructure.utils import generate_instance_id from src.infrastructure.logger import logger from src.infrastructure.config import settings from src.presentation.handler import application_exception_handler, unhandled_exception_handler -from src.presentation.messaging import crypto_transfer_router +from src.presentation.messaging import crypto_transfer_router,sbp_withdrawal_messaging_router from src.presentation.middleware import TraceIDMiddleware, SecurityHeadersMiddleware -from src.presentation.routing import order_router,orders_router,payment_router,payments_router +from src.presentation.routing import order_router,orders_router,payment_router,payments_router,sbp_withdrawal_router security = HTTPBasic() @@ -78,7 +78,9 @@ app.include_router(order_router) app.include_router(orders_router) app.include_router(payment_router) app.include_router(payments_router) +app.include_router(sbp_withdrawal_router) app.include_router(crypto_transfer_router) +app.include_router(sbp_withdrawal_messaging_router) # Added middleware diff --git a/src/presentation/decorators/cache.py b/src/presentation/decorators/cache.py index ed36590..c7b7996 100644 --- a/src/presentation/decorators/cache.py +++ b/src/presentation/decorators/cache.py @@ -1,19 +1,20 @@ from __future__ import annotations - - import functools +import orjson from typing import Any,Awaitable,Callable - from fastapi import Request from fastapi.responses import ORJSONResponse - from src.infrastructure.logger import get_logger from src.presentation.dependencies.cache import get_cache_remote -def cached(*, prefix: str) -> Callable: +def _to_cache_payload(value: Any) -> Any: + if hasattr(value,'model_dump'): + return value.model_dump(mode='json') + return value +def cached(*,prefix: str,ttl: int = 300,user_scoped: bool = True) -> Callable: def decorator(func: Callable[..., Awaitable[Any]]): @functools.wraps(func) async def wrapper(*args: Any, **kwargs: Any) -> Any: @@ -29,21 +30,34 @@ def cached(*, prefix: str) -> Callable: auth = kwargs.get('auth') user_id = getattr(auth, 'user_id', None) if auth else None - if request is None or user_id is None: + if request is None: return await func(*args, **kwargs) - cache_key = f'{prefix}:{user_id}' + if user_scoped: + if user_id is None: + return await func(*args, **kwargs) + cache_key = f'{prefix}:{user_id}' + else: + cache_key = prefix try: cache = get_cache_remote(request) - hit = await cache.get_user(user_id) + hit = await cache.get(cache_key) if hit is not None: logger.debug(f'Cache hit key={cache_key}') - return ORJSONResponse(status_code=200, content=hit) + return ORJSONResponse(status_code=200,content=orjson.loads(hit)) except Exception as e: logger.warning(f'Cache read failed key={cache_key} error={e}') - return await func(*args, **kwargs) + result = await func(*args, **kwargs) + try: + cache = get_cache_remote(request) + payload = _to_cache_payload(result) + await cache.set(cache_key,orjson.dumps(payload).decode(),ttl) + logger.debug(f'Cache set key={cache_key} ttl={ttl}') + except Exception as e: + logger.warning(f'Cache write failed key={cache_key} error={e}') + return result return wrapper return decorator diff --git a/src/presentation/dependencies/commands.py b/src/presentation/dependencies/commands.py index 1997c3a..b485356 100644 --- a/src/presentation/dependencies/commands.py +++ b/src/presentation/dependencies/commands.py @@ -1,14 +1,15 @@ from __future__ import annotations from fastapi import Depends from src.application.abstractions import IUnitOfWork -from src.application.commands import CreateCryptoTransferCompletedCommand,CreateOrderCommand,CreatePaymentCommand,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,ListOrdersCommand,ListPaymentsCommand -from src.application.contracts import ICache,ILogger,IQueueMessanger,IReceipt +from src.application.commands import CreateCryptoTransferCompletedCommand,CreateOrderCommand,CreatePaymentCommand,CreateSbpWithdrawalCommand,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,GetSbpBanksCommand,GetSbpWithdrawalCommand,HandleSbpWithdrawalWalletEventCommand,ListClientOperationsCommand,ListOrdersCommand,ListPaymentsCommand +from src.application.contracts import ICache,ILogger,IMozenSbpService,IQueueMessanger,IReceipt from src.application.contracts.i_itpay_service import IItPayService -from src.application.services import PaymentQuoteService +from src.application.services import PaymentQuoteService,SbpWithdrawalQuoteService from src.infrastructure.cloud_kassir import ClaudeKassirClient from src.infrastructure.cloud_kassir.constants import CLOUD_KASSIR_API_BASE_URL,CLOUD_KASSIR_FAIL_URL,CLOUD_KASSIR_INN,CLOUD_KASSIR_SUCCESS_URL from src.infrastructure.config import settings from src.infrastructure.itpay.client import ItPayClient +from src.infrastructure.mozen import MozenSbpClient from src.presentation.dependencies.cache import get_remote_cache from src.presentation.dependencies.logger import get_logger from src.presentation.dependencies.queue_messanger import get_rabbit @@ -23,6 +24,14 @@ def get_itpay_service() -> IItPayService: ) +def get_mozen_sbp_service() -> IMozenSbpService: + return MozenSbpClient( + api_base_url=settings.MOZEN_API_BASE_URL, + client_id=settings.MOZEN_CLIENT_ID, + park_token=settings.MOZEN_PARK_TOKEN, + ) + + def get_create_order_command( logger: ILogger = Depends(get_logger), unit_of_work: IUnitOfWork = Depends(get_unit_of_work), @@ -45,6 +54,43 @@ def get_payment_quote_service( return PaymentQuoteService(remote_cache=remote_cache,logger=logger) +def get_sbp_withdrawal_quote_service( + logger: ILogger = Depends(get_logger), + remote_cache: ICache = Depends(get_remote_cache), +) -> SbpWithdrawalQuoteService: + return SbpWithdrawalQuoteService(remote_cache=remote_cache,logger=logger) + + +def get_sbp_banks_command( + logger: ILogger = Depends(get_logger), + mozen_sbp_service: IMozenSbpService = Depends(get_mozen_sbp_service), +) -> GetSbpBanksCommand: + return GetSbpBanksCommand(mozen_sbp_service=mozen_sbp_service,logger=logger) + + +def get_create_sbp_withdrawal_command( + logger: ILogger = Depends(get_logger), + unit_of_work: IUnitOfWork = Depends(get_unit_of_work), + queue_messanger: IQueueMessanger = Depends(get_rabbit), + mozen_sbp_service: IMozenSbpService = Depends(get_mozen_sbp_service), + quote_service: SbpWithdrawalQuoteService = Depends(get_sbp_withdrawal_quote_service), +) -> CreateSbpWithdrawalCommand: + return CreateSbpWithdrawalCommand( + unit_of_work=unit_of_work, + logger=logger, + queue_messanger=queue_messanger, + mozen_sbp_service=mozen_sbp_service, + quote_service=quote_service, + ) + + +def get_sbp_withdrawal_command( + logger: ILogger = Depends(get_logger), + unit_of_work: IUnitOfWork = Depends(get_unit_of_work), +) -> GetSbpWithdrawalCommand: + return GetSbpWithdrawalCommand(unit_of_work=unit_of_work,logger=logger) + + def get_payment_config_command( logger: ILogger = Depends(get_logger), payment_quote_service: PaymentQuoteService = Depends(get_payment_quote_service), @@ -80,6 +126,13 @@ def get_list_payments_command( return ListPaymentsCommand(unit_of_work=unit_of_work,logger=logger) +def get_list_client_operations_command( + logger: ILogger = Depends(get_logger), + unit_of_work: IUnitOfWork = Depends(get_unit_of_work), +) -> ListClientOperationsCommand: + return ListClientOperationsCommand(unit_of_work=unit_of_work,logger=logger) + + def get_payment_command( logger: ILogger = Depends(get_logger), unit_of_work: IUnitOfWork = Depends(get_unit_of_work), @@ -125,4 +178,16 @@ def get_crypto_transfer_completed_command( unit_of_work: IUnitOfWork = Depends(get_unit_of_work), receipt: IReceipt = Depends(get_cloud_kassir_receipt), ) -> CreateCryptoTransferCompletedCommand: - return CreateCryptoTransferCompletedCommand(unit_of_work=unit_of_work,receipt=receipt,logger=logger) \ No newline at end of file + return CreateCryptoTransferCompletedCommand(unit_of_work=unit_of_work,receipt=receipt,logger=logger) + + +def get_sbp_withdrawal_wallet_event_command( + logger: ILogger = Depends(get_logger), + unit_of_work: IUnitOfWork = Depends(get_unit_of_work), + mozen_sbp_service: IMozenSbpService = Depends(get_mozen_sbp_service), +) -> HandleSbpWithdrawalWalletEventCommand: + return HandleSbpWithdrawalWalletEventCommand( + unit_of_work=unit_of_work, + logger=logger, + mozen_sbp_service=mozen_sbp_service, + ) \ No newline at end of file diff --git a/src/presentation/messaging/__init__.py b/src/presentation/messaging/__init__.py index d0147d3..1dda968 100644 --- a/src/presentation/messaging/__init__.py +++ b/src/presentation/messaging/__init__.py @@ -1,5 +1,6 @@ from src.presentation.messaging.crypto_transfer import crypto_transfer_router +from src.presentation.messaging.sbp_withdrawal import sbp_withdrawal_messaging_router -__all__=['crypto_transfer_router'] +__all__=['crypto_transfer_router','sbp_withdrawal_messaging_router'] diff --git a/src/presentation/messaging/sbp_withdrawal.py b/src/presentation/messaging/sbp_withdrawal.py new file mode 100644 index 0000000..985c336 --- /dev/null +++ b/src/presentation/messaging/sbp_withdrawal.py @@ -0,0 +1,55 @@ +from fastapi import Depends +from faststream.rabbit import RabbitQueue +from faststream.rabbit.fastapi import RabbitMessage,RabbitRouter +from decimal import Decimal +from pydantic import BaseModel +from src.application.commands import HandleSbpWithdrawalWalletEventCommand +from src.application.contracts import ILogger +from src.infrastructure.config import settings +from src.infrastructure.context_vars import trace_id_var +from src.presentation.dependencies.commands import get_sbp_withdrawal_wallet_event_command +from src.presentation.dependencies.logger import get_logger + + +sbp_withdrawal_messaging_router=RabbitRouter(settings.RABBIT_URL) + + +class SbpWithdrawalWalletEventMessage(BaseModel): + user_id: str + withdraw_id: str + event_type: str + trace_id: str | None = None + received_usdt_amount: Decimal | None = None + tx_hash: str | None = None + error: str | None = None + + +@sbp_withdrawal_messaging_router.subscriber(RabbitQueue(settings.RABBIT_SBP_WITHDRAWAL_WALLET_EVENTS_QUEUE,durable=True)) +async def sbp_withdrawal_wallet_event_handler( + msg_body: SbpWithdrawalWalletEventMessage, + message: RabbitMessage, + command: HandleSbpWithdrawalWalletEventCommand = Depends(get_sbp_withdrawal_wallet_event_command), + logger: ILogger = Depends(get_logger), +) -> None: + headers=message.headers or {} + trace_id=msg_body.trace_id or message.correlation_id or str(headers.get('trace_id') or '') + token=trace_id_var.set(trace_id) + try: + logger.info({ + 'event':'sbp_withdrawal_wallet_event_received', + 'withdraw_id':msg_body.withdraw_id, + 'user_id':msg_body.user_id, + 'wallet_event_type':msg_body.event_type, + 'rabbit_message_id':message.message_id, + 'rabbit_correlation_id':message.correlation_id, + }) + await command( + withdraw_id=msg_body.withdraw_id, + user_id=msg_body.user_id, + event_type=msg_body.event_type, + received_usdt_amount=msg_body.received_usdt_amount, + tx_hash=msg_body.tx_hash, + error=msg_body.error, + ) + finally: + trace_id_var.reset(token) diff --git a/src/presentation/routing/__init__.py b/src/presentation/routing/__init__.py index cc6874e..5cb6f55 100644 --- a/src/presentation/routing/__init__.py +++ b/src/presentation/routing/__init__.py @@ -1 +1,2 @@ -from src.presentation.routing.order import order_router,orders_router,payment_router,payments_router \ No newline at end of file +from src.presentation.routing.order import order_router,orders_router,payment_router,payments_router +from src.presentation.routing.sbp_withdrawal import sbp_withdrawal_router \ No newline at end of file diff --git a/src/presentation/routing/order.py b/src/presentation/routing/order.py index 3569bf3..938bec3 100644 --- a/src/presentation/routing/order.py +++ b/src/presentation/routing/order.py @@ -6,19 +6,19 @@ from fastapi import APIRouter,Depends,Query,Request,WebSocket,WebSocketDisconnec from fastapi.responses import ORJSONResponse from fastapi.security.utils import get_authorization_scheme_param from ulid import ULID -from src.application.commands import CreateOrderCommand,CreatePaymentCommand,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,ListOrdersCommand,ListPaymentsCommand +from src.application.commands import ClientOperationResult,CreateOrderCommand,CreatePaymentCommand,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,ListClientOperationsCommand,ListOrdersCommand from src.application.contracts import IJwtService,ILogger from src.application.domain.dto import AccessTokenPayload,AuthContext -from src.application.domain.entities import OrderEntity,PaymentEntity +from src.application.domain.entities import OrderEntity,PaymentEntity,SbpWithdrawalEntity from src.application.domain.enums import OrderStatus from src.application.domain.exceptions import ApplicationException,ConflictException from src.application.services import PaymentQuote from src.infrastructure.context_vars import trace_id_var from src.presentation.decorators import require_access_token, csrf_protect -from src.presentation.dependencies.commands import get_create_order_command,get_create_payment_command,get_list_orders_command,get_list_payments_command,get_order_command,get_order_status_command,get_payment_command,get_payment_config_command,get_payment_quote_command,get_payment_quote_from_rub_command +from src.presentation.dependencies.commands import get_create_order_command,get_create_payment_command,get_list_client_operations_command,get_list_orders_command,get_order_command,get_order_status_command,get_payment_command,get_payment_config_command,get_payment_quote_command,get_payment_quote_from_rub_command from src.presentation.dependencies.logger import get_logger from src.presentation.dependencies.security import get_jwt_service -from src.presentation.schemas.order import CreateOrder,CreateOrderResponse,ErrorResponse,OrderDetailResponse,OrderPaymentResponse,OrdersResponse,OrderStatusResponse,OrderWithPaymentResponse,PaymentConfigResponse,PaymentDetailResponse,PaymentQuoteResponse,PaymentResponse,PaymentsResponse +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_router = APIRouter(prefix='/order', tags=['orders']) @@ -110,6 +110,40 @@ def _payment_response(payment: PaymentEntity) -> PaymentResponse: ) +def _withdrawal_operation_response(withdrawal: SbpWithdrawalEntity) -> SbpWithdrawalOperationResponse: + return SbpWithdrawalOperationResponse( + id=withdrawal.id, + created_at=withdrawal.created_at.isoformat() if withdrawal.created_at is not None else None, + updated_at=withdrawal.updated_at.isoformat() if withdrawal.updated_at is not None else None, + user_id=withdrawal.user_id, + bank_id=withdrawal.bank_id, + bank_name=withdrawal.bank_name, + phone=withdrawal.phone, + wallet_address=withdrawal.wallet_address, + usdt_amount=str(withdrawal.usdt_amount) if withdrawal.usdt_amount is not None else None, + rub_amount=str(withdrawal.rub_amount) if withdrawal.rub_amount is not None else None, + usdt_exchange_rate=str(withdrawal.usdt_exchange_rate) if withdrawal.usdt_exchange_rate is not None else None, + service_fee_rate=str(withdrawal.service_fee_rate) if withdrawal.service_fee_rate is not None else None, + service_fee_usdt=str(withdrawal.service_fee_usdt) if withdrawal.service_fee_usdt is not None else None, + mozen_order_id=withdrawal.mozen_order_id, + trace_id=withdrawal.trace_id, + status=withdrawal.status, + provider_error=withdrawal.provider_error, + usdt_received_at=withdrawal.usdt_received_at.isoformat() if withdrawal.usdt_received_at is not None else None, + payout_created_at=withdrawal.payout_created_at.isoformat() if withdrawal.payout_created_at is not None else None, + payout_completed_at=withdrawal.payout_completed_at.isoformat() if withdrawal.payout_completed_at is not None else None, + ) + + +def _client_operation_response(operation: ClientOperationResult) -> ClientOperationResponse: + return ClientOperationResponse( + type=operation.type, + created_at=operation.created_at.isoformat(), + payment=_payment_response(operation.payment) if operation.payment is not None else None, + withdrawal=_withdrawal_operation_response(operation.withdrawal) if operation.withdrawal is not None else None, + ) + + def _order_detail_response(order: OrderEntity, payment: PaymentEntity | None) -> OrderDetailResponse: return OrderDetailResponse( status_code=200, @@ -266,7 +300,7 @@ async def payment_list_orders( @payment_router.get( '/payments', - response_model=PaymentsResponse, + response_model=ClientOperationsResponse, status_code=200, responses=ERROR_RESPONSES, ) @@ -276,12 +310,12 @@ async def payment_list_payments( limit: int = Query(default=20, ge=1, le=100), offset: int = Query(default=0, ge=0), auth: AuthContext = Depends(require_access_token), - command: ListPaymentsCommand = Depends(get_list_payments_command), -) -> PaymentsResponse: - payments = await command(user_id=auth.user_id,limit=limit,offset=offset) - return PaymentsResponse( + command: ListClientOperationsCommand = Depends(get_list_client_operations_command), +) -> ClientOperationsResponse: + operations = await command(user_id=auth.user_id,limit=limit,offset=offset) + return ClientOperationsResponse( status_code=200, - payments=[_payment_response(payment) for payment in payments], + operations=[_client_operation_response(operation) for operation in operations], limit=limit, offset=offset, ) @@ -325,7 +359,7 @@ async def list_orders( @payments_router.get( '/payments', - response_model=PaymentsResponse, + response_model=ClientOperationsResponse, status_code=200, responses=ERROR_RESPONSES, ) @@ -335,12 +369,12 @@ async def list_payments( limit: int = Query(default=20, ge=1, le=100), offset: int = Query(default=0, ge=0), auth: AuthContext = Depends(require_access_token), - command: ListPaymentsCommand = Depends(get_list_payments_command), -) -> PaymentsResponse: - payments = await command(user_id=auth.user_id,limit=limit,offset=offset) - return PaymentsResponse( + command: ListClientOperationsCommand = Depends(get_list_client_operations_command), +) -> ClientOperationsResponse: + operations = await command(user_id=auth.user_id,limit=limit,offset=offset) + return ClientOperationsResponse( status_code=200, - payments=[_payment_response(payment) for payment in payments], + operations=[_client_operation_response(operation) for operation in operations], limit=limit, offset=offset, ) diff --git a/src/presentation/routing/sbp_withdrawal.py b/src/presentation/routing/sbp_withdrawal.py new file mode 100644 index 0000000..b75a58a --- /dev/null +++ b/src/presentation/routing/sbp_withdrawal.py @@ -0,0 +1,107 @@ +from fastapi import APIRouter,Depends,Request +from src.application.commands import CreateSbpWithdrawalCommand,GetSbpBanksCommand,GetSbpWithdrawalCommand +from src.application.domain.dto import AuthContext +from src.application.domain.entities import SbpWithdrawalEntity +from src.presentation.decorators import cached,csrf_protect,require_access_token +from src.presentation.dependencies.commands import get_create_sbp_withdrawal_command,get_sbp_banks_command,get_sbp_withdrawal_command +from src.presentation.schemas.order import ErrorResponse +from src.presentation.schemas.sbp_withdrawal import CreateSbpWithdrawal,CreateSbpWithdrawalResponse,SbpBankResponse,SbpBanksResponse,SbpWithdrawalDetailResponse,SbpWithdrawalResponse + + +sbp_withdrawal_router=APIRouter(prefix='/sbp-payout',tags=['sbp payouts']) + + +ERROR_RESPONSES={ + 400:{'model':ErrorResponse,'description':'Bad Request'}, + 401:{'model':ErrorResponse,'description':'Unauthorized'}, + 403:{'model':ErrorResponse,'description':'Forbidden'}, + 404:{'model':ErrorResponse,'description':'Not Found'}, + 409:{'model':ErrorResponse,'description':'Conflict'}, + 422:{'model':ErrorResponse,'description':'Validation Error'}, + 429:{'model':ErrorResponse,'description':'Too Many Requests'}, + 500:{'model':ErrorResponse,'description':'Internal Server Error'}, + 502:{'model':ErrorResponse,'description':'Bad Gateway'}, + 503:{'model':ErrorResponse,'description':'Service Unavailable'}, +} + + +def _withdrawal_response(withdrawal: SbpWithdrawalEntity) -> SbpWithdrawalResponse: + return SbpWithdrawalResponse( + id=withdrawal.id, + created_at=withdrawal.created_at.isoformat() if withdrawal.created_at is not None else None, + updated_at=withdrawal.updated_at.isoformat() if withdrawal.updated_at is not None else None, + user_id=withdrawal.user_id, + bank_id=withdrawal.bank_id, + bank_name=withdrawal.bank_name, + phone=withdrawal.phone, + wallet_address=withdrawal.wallet_address, + usdt_amount=str(withdrawal.usdt_amount) if withdrawal.usdt_amount is not None else None, + rub_amount=str(withdrawal.rub_amount) if withdrawal.rub_amount is not None else None, + usdt_exchange_rate=str(withdrawal.usdt_exchange_rate) if withdrawal.usdt_exchange_rate is not None else None, + service_fee_rate=str(withdrawal.service_fee_rate) if withdrawal.service_fee_rate is not None else None, + service_fee_usdt=str(withdrawal.service_fee_usdt) if withdrawal.service_fee_usdt is not None else None, + mozen_order_id=withdrawal.mozen_order_id, + trace_id=withdrawal.trace_id, + status=withdrawal.status, + provider_error=withdrawal.provider_error, + usdt_received_at=withdrawal.usdt_received_at.isoformat() if withdrawal.usdt_received_at is not None else None, + payout_created_at=withdrawal.payout_created_at.isoformat() if withdrawal.payout_created_at is not None else None, + payout_completed_at=withdrawal.payout_completed_at.isoformat() if withdrawal.payout_completed_at is not None else None, + ) + + +@sbp_withdrawal_router.get( + '/banks', + response_model=SbpBanksResponse, + status_code=200, + responses=ERROR_RESPONSES, +) +@cached(prefix='sbp:payout:banks',ttl=3600,user_scoped=False) +async def sbp_banks( + request: Request, + command: GetSbpBanksCommand = Depends(get_sbp_banks_command), +) -> SbpBanksResponse: + banks=await command() + return SbpBanksResponse( + status_code=200, + banks=[SbpBankResponse(id=bank.id,name=bank.name) for bank in banks], + ) + + +@sbp_withdrawal_router.post( + '/withdrawals', + response_model=CreateSbpWithdrawalResponse, + status_code=201, + responses=ERROR_RESPONSES, +) +@csrf_protect() +async def create_sbp_withdrawal( + request: Request, + payload: CreateSbpWithdrawal, + auth: AuthContext = Depends(require_access_token), + command: CreateSbpWithdrawalCommand = Depends(get_create_sbp_withdrawal_command), +) -> CreateSbpWithdrawalResponse: + withdrawal=await command( + user_id=auth.user_id, + bank_id=payload.bank_id, + phone=payload.phone, + usdt_amount=payload.usdt_amount, + ) + return CreateSbpWithdrawalResponse(status_code=201,withdrawal=_withdrawal_response(withdrawal)) + + +@sbp_withdrawal_router.get( + '/withdrawals/{withdraw_id}', + response_model=SbpWithdrawalDetailResponse, + status_code=200, + responses=ERROR_RESPONSES, +) +@csrf_protect() +async def sbp_withdrawal_detail( + request: Request, + withdraw_id: str, + auth: AuthContext = Depends(require_access_token), + command: GetSbpWithdrawalCommand = Depends(get_sbp_withdrawal_command), +) -> SbpWithdrawalDetailResponse: + withdrawal=await command(withdraw_id=withdraw_id,user_id=auth.user_id) + return SbpWithdrawalDetailResponse(status_code=200,withdrawal=_withdrawal_response(withdrawal)) diff --git a/src/presentation/schemas/order.py b/src/presentation/schemas/order.py index 12bbacd..691b972 100644 --- a/src/presentation/schemas/order.py +++ b/src/presentation/schemas/order.py @@ -1,6 +1,6 @@ from decimal import Decimal from pydantic import BaseModel,Field -from src.application.domain.enums import OrderStatus,PaymentStatus +from src.application.domain.enums import OrderStatus,PaymentStatus,SbpWithdrawalStatus class CreateOrder(BaseModel): @@ -51,6 +51,36 @@ class PaymentResponse(BaseModel): expired_date: str | None = None +class SbpWithdrawalOperationResponse(BaseModel): + id: str | None = None + created_at: str | None = None + updated_at: str | None = None + user_id: str | None = None + bank_id: str | None = None + bank_name: str | None = None + phone: str | None = None + wallet_address: str | None = None + usdt_amount: str | None = None + rub_amount: str | None = None + usdt_exchange_rate: str | None = None + service_fee_rate: str | None = None + service_fee_usdt: str | None = None + mozen_order_id: str | None = None + trace_id: str | None = None + status: SbpWithdrawalStatus | None = None + provider_error: str | None = None + usdt_received_at: str | None = None + payout_created_at: str | None = None + payout_completed_at: str | None = None + + +class ClientOperationResponse(BaseModel): + type: str + created_at: str + payment: PaymentResponse | None = None + withdrawal: SbpWithdrawalOperationResponse | None = None + + class CreateOrderResponse(BaseModel): status_code: int order: OrderPaymentResponse @@ -100,6 +130,13 @@ class PaymentsResponse(BaseModel): offset: int +class ClientOperationsResponse(BaseModel): + status_code: int + operations: list[ClientOperationResponse] + limit: int + offset: int + + class PaymentDetailResponse(BaseModel): status_code: int payment: PaymentResponse diff --git a/src/presentation/schemas/sbp_withdrawal.py b/src/presentation/schemas/sbp_withdrawal.py new file mode 100644 index 0000000..281863b --- /dev/null +++ b/src/presentation/schemas/sbp_withdrawal.py @@ -0,0 +1,52 @@ +from decimal import Decimal +from pydantic import BaseModel,Field +from src.application.domain.enums import SbpWithdrawalStatus + + +class SbpBankResponse(BaseModel): + id: str + name: str + + +class SbpBanksResponse(BaseModel): + status_code: int + banks: list[SbpBankResponse] + + +class CreateSbpWithdrawal(BaseModel): + bank_id: str = Field(min_length=1,max_length=32) + phone: str = Field(min_length=5,max_length=32) + usdt_amount: Decimal = Field(ge=Decimal('100'),le=Decimal('5000'),decimal_places=2,max_digits=20) + + +class SbpWithdrawalResponse(BaseModel): + id: str | None = None + created_at: str | None = None + updated_at: str | None = None + user_id: str | None = None + bank_id: str | None = None + bank_name: str | None = None + phone: str | None = None + wallet_address: str | None = None + usdt_amount: str | None = None + rub_amount: str | None = None + usdt_exchange_rate: str | None = None + service_fee_rate: str | None = None + service_fee_usdt: str | None = None + mozen_order_id: str | None = None + trace_id: str | None = None + status: SbpWithdrawalStatus | None = None + provider_error: str | None = None + usdt_received_at: str | None = None + payout_created_at: str | None = None + payout_completed_at: str | None = None + + +class CreateSbpWithdrawalResponse(BaseModel): + status_code: int + withdrawal: SbpWithdrawalResponse + + +class SbpWithdrawalDetailResponse(BaseModel): + status_code: int + withdrawal: SbpWithdrawalResponse