feat: add notifies to rabbitmq

src/application/commands/purchase_request_commands.py

@@ -1,19 +1,25 @@
 from __future__ import annotations
 
+from datetime import datetime, timezone
 from decimal import Decimal
+from ulid import ULID
 
+from src.application.domain.exceptions.application_exceptions import ApplicationException
+from src.infrastructure.config import settings
+from src.infrastructure.context_vars import trace_id_var
 from src.application.abstractions import IUnitOfWork
 from src.application.commands.legal_entity_guard import require_legal_entity
-from src.application.contracts import ILogger
+from src.application.contracts import ILogger, IQueueMessanger
 from src.application.domain.entities.purchase_request import PurchaseRequestEntity
 from src.application.domain.exceptions import NotFoundException
 from src.infrastructure.database.decorators import transactional
 
 
 class CreatePurchaseRequestCommand:
-    def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger):
+    def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger, messanger: IQueueMessanger):
         self._unit_of_work = unit_of_work
         self._logger = logger
+        self._messanger = messanger
 
     @transactional
     async def __call__(
@@ -33,7 +39,41 @@ class CreatePurchaseRequestCommand:
             target_wallet_chain=target_wallet_chain,
             target_wallet_address=target_wallet_address,
         )
-        self._logger.info(f'Purchase request created id={item.id} organization_id={legal_entity.id}')
+        trace_id = trace_id_var.get()
+        if not trace_id or trace_id == 'N/A':
+            trace_id = None
+
+        message_id = str(ULID())
+        now = datetime.now(timezone.utc).isoformat()
+        metadata = {
+            'trace_id': trace_id,
+            'timestamp': now,
+            'message_id': message_id,
+        }
+        payload = {
+            'id_client': legal_entity.id,
+            'usdt_amount': usdt_amount,
+        }
+        message = {
+            'event': 'purchase_created',
+            'payload': payload,
+            'metadata': metadata,
+        }
+        self._logger.info(f'Create purchase request for organization_id={legal_entity.id } with usdt_amount={usdt_amount}')
+        try:
+            await self._messanger.publish_to_queue(
+                queue=settings.RABBIT_TELEGRAM_NOTIFY_QUEUE,
+                message=message,
+                persist=True,
+                correlation_id=trace_id,
+                message_id=message_id,
+                headers={'trace_id': trace_id} if trace_id else None,
+            )
+        except Exception as exception:
+            self._logger.error(f'Failed to publish purchase request notification for organization_id={legal_entity.id}: {str(exception)}')
+            raise ApplicationException(503, 'Temporary error. Please try again.')
+
+        self._logger.info(f'Purchase request created id= organization_id={legal_entity.id }')
         return item
 
 

src/application/commands/upload_docs_button.py

@@ -0,0 +1,57 @@
+# Пустышка на будущее, когда юр лицо загрузит документы, нужно будет отправить уведу в телегу,
+#  что юр лицо загрузило документы и их нужно проверить
+
+from __future__ import annotations
+from datetime import datetime, timezone
+from ulid import ULID
+
+from src.application.domain.exceptions.application_exceptions import ApplicationException
+from src.infrastructure.config import settings
+from src.infrastructure.context_vars import trace_id_var
+from src.application.abstractions import IUnitOfWork
+from src.application.commands.legal_entity_guard import require_legal_entity
+from src.infrastructure.database.decorators import transactional
+from src.application.contracts import ILogger, IQueueMessanger
+
+class UploadDocsCommand:
+    def __init__(self, unit_of_work: IUnitOfWork, logger: ILogger, messanger: IQueueMessanger):
+        self._unit_of_work = unit_of_work
+        self._logger = logger
+        self._messanger = messanger
+
+    @transactional
+    async def __call__(
+        self,
+        user_id: str,
+        *,
+        comment: str | None = None,
+    ) -> None:
+        legal_entity = await require_legal_entity(user_id, self._unit_of_work)
+        message_id = str(ULID())
+        now = datetime.now(timezone.utc).isoformat()
+        metadata = {
+            'trace_id': trace_id_var.get(),
+            'timestamp': now,
+            'message_id': message_id,
+        }
+        payload = {
+            'id_client': legal_entity.id,
+        }
+        message = {
+            'event': 'docs_upload',
+            'payload': payload,
+            'metadata': metadata,
+        }
+        self._logger.info(f'Upload docs for organization_id={legal_entity.id}')
+        try:
+            await self._messanger.publish_to_queue(
+                queue=settings.RABBIT_TELEGRAM_NOTIFY_QUEUE,
+                message=message,
+                persist=True,
+                correlation_id=trace_id_var.get(),
+                message_id=message_id,
+                headers={'trace_id': trace_id_var.get()} if trace_id_var.get() else None,
+            )
+        except Exception as exception:
+            self._logger.error(f'Failed to publish info notification for organization_id={legal_entity.id}: {str(exception)}')
+            raise ApplicationException(503, 'Temporary error. Please try again.')

src/application/contracts/__init__.py

@@ -1,3 +1,4 @@
 from src.application.contracts.i_logger import ILogger
 from src.application.contracts.i_jwt_service import IJwtService
 from src.application.contracts.i_csrf_service import ICsrfService
+from src.application.contracts.i_queue_messanger import IQueueMessanger

src/application/contracts/i_queue_messanger.py

@@ -0,0 +1,40 @@
+from abc import ABC, abstractmethod
+from typing import Mapping, Any
+
+
+class IQueueMessanger(ABC):
+
+    @abstractmethod
+    async def connect(self) -> None:
+        raise NotImplementedError
+    
+    @abstractmethod
+    async def close(self) -> None:
+        raise NotImplementedError
+    
+    @abstractmethod
+    async def publish_to_queue(
+            self,
+            queue: str,
+            message: Any,
+            *,
+            persist: bool = True,
+            headers: Mapping[str, Any] | None = None,
+            correlation_id: str | None = None,
+            message_id: str | None = None,
+    ) -> None:
+        raise NotImplementedError
+    
+    @abstractmethod
+    async def publish(
+            self,
+            message: Any,
+            *,
+            exchange: str,
+            routing_key: str,
+            persist: bool = True,
+            headers: Mapping[str, Any] | None = None,
+            correlation_id: str | None = None,
+            message_id: str | None = None,
+    ) -> None:
+        raise NotImplementedError

src/infrastructure/config/settings.py

@@ -34,6 +34,7 @@ class Settings(BaseSettings):
     VAULT_NAMESPACE: str | None = None
     VAULT_MOUNT_POINT: str = 'dev-secrets'
     VAULT_DATABASE_SECRET_PATH: str = 'database'
+    VAULT_RABBIT_SECRET_PATH: str = 'rabbitmq'
     VAULT_CSRF_SECRET_PATH: str = 'csrf'
     VAULT_DOCS_SECRET_PATH: str = 'docs'
     VAULT_JWT_KID_PATH: str = 'jwt/kid'
@@ -74,6 +75,16 @@ class Settings(BaseSettings):
     JWT_ALGORITHM: str = 'RS256'
     JWT_KEYS_REFRESH_SECONDS: int = 3600
 
+    RABBIT_HOST: str = 'localhost'
+    RABBIT_PORT: int = 5672
+    RABBIT_USER: str = 'guest'
+    RABBIT_PASSWORD: str = 'guest'
+    RABBIT_VHOST: str = '/'
+    RABBIT_PUBLISH_PERSIST: bool = True
+    RABBIT_CONNECT_TIMEOUT: int = 5
+    RABBIT_EMAIL_CODE_QUEUE: str = 'email.verification_code'
+    RABBIT_TELEGRAM_NOTIFY_QUEUE: str = "telegram.notify"
+
     LOG_LEVEL: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] = 'INFO'
     LOG_FORMAT: Literal['JSON', 'TEXT'] = 'JSON'
 
@@ -120,6 +131,19 @@ class Settings(BaseSettings):
             object.__setattr__(self, 'DATABASE_USER', str(kv(db, 'USER', 'user')))
         if kv(db, 'PASSWORD', 'password') is not None:
             object.__setattr__(self, 'DATABASE_PASSWORD', str(kv(db, 'PASSWORD', 'password')))
+        
+        rabbit = client.read_secret_optional(self.VAULT_RABBIT_SECRET_PATH)
+        if rabbit:
+            if kv(rabbit, 'HOST', 'host') is not None:
+                object.__setattr__(self, 'RABBIT_HOST', str(kv(rabbit, 'HOST', 'host')))
+            if kv(rabbit, 'PORT', 'port') is not None:
+                object.__setattr__(self, 'RABBIT_PORT', _as_int(kv(rabbit, 'PORT', 'port'), self.RABBIT_PORT))
+            if kv(rabbit, 'USER', 'user') is not None:
+                object.__setattr__(self, 'RABBIT_USER', str(kv(rabbit, 'USER', 'user')))
+            if kv(rabbit, 'PASSWORD', 'password') is not None:
+                object.__setattr__(self, 'RABBIT_PASSWORD', str(kv(rabbit, 'PASSWORD', 'password')))
+            if kv(rabbit, 'VHOST', 'vhost') is not None:
+                object.__setattr__(self, 'RABBIT_VHOST', str(kv(rabbit, 'VHOST', 'vhost')))
 
         csrf = client.read_secret_optional(self.VAULT_CSRF_SECRET_PATH)
         if csrf and kv(csrf, 'KEY', 'key') is not None:
@@ -172,6 +196,11 @@ class Settings(BaseSettings):
         quoted_user = quote(user, safe='')
         quoted_password = quote(password, safe='')
         return f'postgresql+asyncpg://{quoted_user}:{quoted_password}@{host}:{port}/{name}'
+    
+    @property
+    def RABBIT_URL(self) -> str:
+        vhost = '%2F' if self.RABBIT_VHOST == '/' else self.RABBIT_VHOST.lstrip('/')
+        return f'amqp://{self.RABBIT_USER}:{self.RABBIT_PASSWORD}@{self.RABBIT_HOST}:{self.RABBIT_PORT}/{vhost}'
 
     @property
     def EXCLUDED_PATHS(self) -> List[str]:

src/infrastructure/messanger/__init__.py

@@ -0,0 +1 @@
+from src.infrastructure.messanger.rabbit_client import RabbitClient

src/infrastructure/messanger/rabbit_client.py

@@ -0,0 +1,73 @@
+from typing import Any, Mapping
+from faststream.rabbit import RabbitBroker
+from src.application.contracts import IQueueMessanger
+from src.infrastructure.config import settings
+
+
+class RabbitClient(IQueueMessanger):
+    def __init__(self) -> None:
+        self._broker = RabbitBroker(
+            settings.RABBIT_URL,
+        )
+        self._connected = False
+
+    async def connect(self) -> None:
+        if self._connected:
+            return
+        await self._broker.connect()
+        self._connected = True
+
+    async def close(self) -> None:
+        if not self._connected:
+            return
+        await self._broker.close()
+        self._connected = False
+
+    async def _ensure_connected(self) -> None:
+        if not self._connected:
+            await self.connect()
+
+    async def publish_to_queue(
+        self,
+        queue: str,
+        message: Any,
+        *,
+        persist: bool | None = None,
+        headers: Mapping[str, Any] | None = None,
+        correlation_id: str | None = None,
+        message_id: str | None = None,
+    ) -> None:
+        print("check2")
+        await self._ensure_connected()
+        print("check3")
+        await self._broker.publish(
+            message,
+            queue=queue,
+            persist=settings.RABBIT_PUBLISH_PERSIST if persist is None else persist,
+            headers=headers,
+            correlation_id=correlation_id,
+            message_id=message_id,
+        )
+
+    async def publish(
+        self,
+        message: Any,
+        *,
+        exchange: str,
+        routing_key: str,
+        persist: bool | None = None,
+        headers: Mapping[str, Any] | None = None,
+        correlation_id: str | None = None,
+        message_id: str | None = None,
+    ) -> None:
+        await self._ensure_connected()
+
+        await self._broker.publish(
+            message,
+            exchange=exchange,
+            routing_key=routing_key,
+            persist=settings.RABBIT_PUBLISH_PERSIST if persist is None else persist,
+            headers=headers,
+            correlation_id=correlation_id,
+            message_id=message_id,
+        )

src/presentation/dependencies/commands.py

@@ -10,16 +10,18 @@ from src.application.commands import (
     ListOrganizationsCommand,
     ListPurchaseRequestsCommand,
 )
-from src.application.contracts import ILogger
+from src.application.contracts import ILogger, IQueueMessanger
 from src.presentation.dependencies.logger import get_logger
 from src.presentation.dependencies.unit_of_work import get_unit_of_work
+from src.presentation.dependencies.messanger import get_queue_messanger
 
 
 def get_create_purchase_request_command(
     logger: ILogger = Depends(get_logger),
     unit_of_work: IUnitOfWork = Depends(get_unit_of_work),
+    messanger: IQueueMessanger = Depends(get_queue_messanger),
 ) -> CreatePurchaseRequestCommand:
-    return CreatePurchaseRequestCommand(unit_of_work=unit_of_work, logger=logger)
+    return CreatePurchaseRequestCommand(unit_of_work=unit_of_work, logger=logger, messanger=messanger)
 
 
 def get_list_purchase_requests_command(

src/presentation/dependencies/messanger.py

@@ -0,0 +1,7 @@
+from functools import lru_cache
+from src.application.contracts import IQueueMessanger
+from src.infrastructure.messanger import RabbitClient
+
+@lru_cache
+def get_queue_messanger() -> IQueueMessanger:
+    return RabbitClient()