127 lines
4.9 KiB
Python
127 lines
4.9 KiB
Python
import secrets
|
|
from datetime import datetime, timezone
|
|
from ulid import ULID
|
|
from src.application.abstractions import IUnitOfWork
|
|
from src.application.contracts import IHashService, ICache, ILogger, IQueueMessanger
|
|
from src.application.domain.exceptions import ApplicationException
|
|
from src.infrastructure.config import settings
|
|
from src.infrastructure.context_vars import trace_id_var
|
|
from src.infrastructure.database.decorators import transactional
|
|
|
|
|
|
class UpdateBankDetailsStartCommand:
|
|
def __init__(
|
|
self,
|
|
hash_service: IHashService,
|
|
cache: ICache,
|
|
unit_of_work: IUnitOfWork,
|
|
logger: ILogger,
|
|
messanger: IQueueMessanger,
|
|
):
|
|
self._hash_service = hash_service
|
|
self._unit_of_work = unit_of_work
|
|
self._cache = cache
|
|
self._logger = logger
|
|
self._messanger = messanger
|
|
|
|
@transactional
|
|
async def __call__(self, user_id: str) -> bool:
|
|
TTL = 300
|
|
LOCK_TTL = 30
|
|
MAX_ATTEMPTS = 20
|
|
|
|
USER_PREFIX = 'bank_details:user:'
|
|
CODE_PREFIX = 'bank_details:code:'
|
|
LOCK_PREFIX = 'bank_details:lock:'
|
|
|
|
user = await self._unit_of_work.user_repository.get_user_by_id(user_id=user_id)
|
|
|
|
if not user.email:
|
|
self._logger.warning(f'User {user_id} does not have an email address')
|
|
raise ApplicationException(status_code=404, message=f'User {user_id} does not have an email address')
|
|
|
|
trace_id = trace_id_var.get()
|
|
if not trace_id or trace_id == 'N/A':
|
|
trace_id = None
|
|
|
|
lock_key = f'{LOCK_PREFIX}{user_id}'
|
|
locked = await self._cache.set_nx(lock_key, '1', ttl=LOCK_TTL)
|
|
if not locked:
|
|
self._logger.info(f'Bank details update throttled by lock (user_id={user_id})')
|
|
raise ApplicationException(429, 'Too many requests. Please wait.')
|
|
|
|
try:
|
|
user_key = f'{USER_PREFIX}{user_id}'
|
|
|
|
existing = await self._cache.get(user_key)
|
|
if existing:
|
|
self._logger.info(f'Bank details update denied: code already exists for user_id={user_id}')
|
|
raise ApplicationException(429, 'Code already sent. Please wait before retrying.')
|
|
|
|
for _ in range(MAX_ATTEMPTS):
|
|
code = f'{secrets.randbelow(1_000_000):06d}'
|
|
code_key = f'{CODE_PREFIX}{code}'
|
|
|
|
code_hash = await self._hash_service.hash(code)
|
|
|
|
reserved = await self._cache.set_nx(code_key, user_id, ttl=TTL)
|
|
if not reserved:
|
|
continue
|
|
|
|
saved = await self._cache.set(user_key, code_hash, ttl=TTL)
|
|
if not saved:
|
|
await self._cache.delete(code_key)
|
|
self._logger.error(f'Bank details update failed: cannot save code hash for user_id={user_id}')
|
|
raise ApplicationException(503, 'Temporary error. Please try again.')
|
|
|
|
message_id = str(ULID())
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
|
|
metadata = {
|
|
'trace_id': trace_id,
|
|
'source': 'user-service',
|
|
'timestamp': now,
|
|
'message_id': message_id,
|
|
}
|
|
|
|
payload = {
|
|
'email': user.email,
|
|
'code': code,
|
|
'ttl_seconds': TTL,
|
|
}
|
|
|
|
message = {
|
|
'event': 'bank_details_update',
|
|
'payload': payload,
|
|
'metadata': metadata,
|
|
}
|
|
|
|
self._logger.info(f'Bank details update code created for user_id={user_id}')
|
|
|
|
try:
|
|
await self._messanger.publish_to_queue(
|
|
queue=settings.RABBIT_EMAIL_CODE_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:
|
|
try:
|
|
await self._cache.delete(user_key)
|
|
await self._cache.delete(code_key)
|
|
except Exception as rollback_err:
|
|
self._logger.error(f'Publish failed and rollback cache failed for user_id={user_id}: {str(rollback_err)}')
|
|
|
|
self._logger.error(f'Failed to publish bank details update email for user_id={user_id}: {str(exception)}')
|
|
raise ApplicationException(503, 'Temporary error. Please try again.')
|
|
|
|
return True
|
|
|
|
self._logger.error(f'Bank details update failed: code space exhausted for user_id={user_id}')
|
|
raise ApplicationException(503, 'Temporary error. Please try again.')
|
|
|
|
finally:
|
|
await self._cache.delete(lock_key)
|