feat(infra): add persistence and SMS adapters

src/gapido_auth/infrastructure/__init__.py

@@ -0,0 +1,2 @@
+"""Infrastructure adapters."""
+

src/gapido_auth/infrastructure/container.py

@@ -0,0 +1,60 @@
+from dataclasses import dataclass
+from typing import Any
+
+from aio_pika.abc import AbstractChannel
+from motor.motor_asyncio import AsyncIOMotorClient
+from redis.asyncio import Redis
+
+from gapido_auth.application.auth_service import AuthConfig, AuthService
+from gapido_auth.application.security import JwtTokenCodec
+from gapido_auth.config import Settings
+from gapido_auth.infrastructure.mongo_repositories import (
+    MongoRefreshSessionRepository,
+    MongoUserRepository,
+)
+from gapido_auth.infrastructure.rabbitmq import RabbitMqSmsPublisher
+from gapido_auth.infrastructure.redis_otp_store import RedisOtpStore
+
+
+@dataclass(slots=True)
+class AppContainer:
+    auth_service: AuthService
+    mongo_client: AsyncIOMotorClient[Any]
+    redis: Redis
+
+
+async def build_auth_service(settings: Settings, rabbitmq_channel: AbstractChannel) -> AppContainer:
+    mongo_client: AsyncIOMotorClient[Any] = AsyncIOMotorClient(settings.mongo_uri)
+    db = mongo_client[settings.mongo_db_name]
+    users = MongoUserRepository(db)
+    refresh_sessions = MongoRefreshSessionRepository(db)
+    await users.create_indexes()
+    await refresh_sessions.create_indexes()
+    await users.ensure_admin_user(settings.admin_mobile)
+
+    redis = Redis.from_url(settings.redis_url, decode_responses=False)
+    otp_store = RedisOtpStore(redis)
+    sms_publisher = RabbitMqSmsPublisher(rabbitmq_channel)
+    token_codec = JwtTokenCodec(
+        secret_key=settings.jwt_secret_key,
+        issuer=settings.jwt_issuer,
+        access_ttl_seconds=settings.access_token_ttl_seconds,
+    )
+    auth_service = AuthService(
+        users=users,
+        refresh_sessions=refresh_sessions,
+        otp_store=otp_store,
+        sms_publisher=sms_publisher,
+        token_codec=token_codec,
+        config=AuthConfig(
+            otp_secret=settings.jwt_secret_key,
+            otp_ttl_seconds=settings.otp_ttl_seconds,
+            otp_max_attempts=settings.otp_max_attempts,
+            otp_request_limit=settings.otp_request_limit,
+            otp_request_window_seconds=settings.otp_request_window_seconds,
+            refresh_token_ttl_seconds=settings.refresh_token_ttl_seconds,
+            sms_template=settings.kavenegar_login_template,
+        ),
+    )
+    return AppContainer(auth_service=auth_service, mongo_client=mongo_client, redis=redis)
+

src/gapido_auth/infrastructure/kavenegar_client.py

@@ -0,0 +1,45 @@
+import logging
+from typing import Any
+
+import httpx
+
+from gapido_auth.domain.errors import ExternalServiceError
+from gapido_auth.domain.ports import SmsClient
+
+logger = logging.getLogger(__name__)
+
+
+class KavenegarSmsClient(SmsClient):
+    def __init__(
+        self,
+        api_key: str,
+        timeout_seconds: float = 10.0,
+        transport: httpx.AsyncBaseTransport | None = None,
+    ) -> None:
+        self._api_key = api_key
+        self._timeout_seconds = timeout_seconds
+        self._transport = transport
+
+    async def send_otp(self, mobile: str, code: str, template: str) -> None:
+        url = f"https://api.kavenegar.com/v1/{self._api_key}/verify/lookup.json"
+        payload = {"receptor": mobile, "template": template, "token": code, "type": "sms"}
+        try:
+            async with httpx.AsyncClient(
+                timeout=self._timeout_seconds, transport=self._transport
+            ) as client:
+                response = await client.post(url, data=payload)
+        except httpx.HTTPError as exc:
+            logger.warning("Kavenegar network error for mobile=%s", mobile, exc_info=True)
+            raise ExternalServiceError("Kavenegar network error") from exc
+
+        if response.status_code != 200:
+            logger.warning("Kavenegar HTTP error status=%s mobile=%s", response.status_code, mobile)
+            raise ExternalServiceError("Kavenegar HTTP error")
+
+        data: dict[str, Any] = response.json()
+        status = data.get("return", {}).get("status")
+        if status != 200:
+            logger.warning("Kavenegar API error status=%s mobile=%s", status, mobile)
+            raise ExternalServiceError("Kavenegar API error")
+
+        logger.info("OTP SMS sent successfully to mobile=%s", mobile)

src/gapido_auth/infrastructure/mongo_repositories.py

@@ -0,0 +1,135 @@
+from datetime import UTC, datetime
+from typing import Any
+
+from bson import ObjectId
+from motor.motor_asyncio import AsyncIOMotorDatabase
+from pymongo import ASCENDING
+
+from gapido_auth.domain.entities import RefreshSession, Role, User
+from gapido_auth.domain.ports import RefreshSessionRepository, UserRepository
+
+
+def _now() -> datetime:
+    return datetime.now(UTC)
+
+
+def _user_from_doc(doc: dict[str, Any]) -> User:
+    return User(
+        id=str(doc["_id"]),
+        mobile=str(doc["mobile"]),
+        role=Role(str(doc["role"])),
+        is_active=bool(doc["is_active"]),
+        created_at=doc["created_at"],
+        updated_at=doc["updated_at"],
+    )
+
+
+def _session_from_doc(doc: dict[str, Any]) -> RefreshSession:
+    return RefreshSession(
+        id=str(doc["_id"]),
+        user_id=str(doc["user_id"]),
+        token_hash=str(doc["token_hash"]),
+        expires_at=doc["expires_at"],
+        revoked_at=doc.get("revoked_at"),
+        replaced_by_hash=doc.get("replaced_by_hash"),
+        created_at=doc["created_at"],
+    )
+
+
+class MongoUserRepository(UserRepository):
+    def __init__(self, db: AsyncIOMotorDatabase[Any]) -> None:
+        self._collection = db.users
+
+    async def create_indexes(self) -> None:
+        await self._collection.create_index([("mobile", ASCENDING)], unique=True)
+        await self._collection.create_index([("role", ASCENDING)])
+
+    async def get_by_id(self, user_id: str) -> User | None:
+        if not ObjectId.is_valid(user_id):
+            return None
+        doc = await self._collection.find_one({"_id": ObjectId(user_id)})
+        return _user_from_doc(doc) if doc else None
+
+    async def get_by_mobile(self, mobile: str) -> User | None:
+        doc = await self._collection.find_one({"mobile": mobile})
+        return _user_from_doc(doc) if doc else None
+
+    async def get_or_create_user(self, mobile: str, role: Role = Role.USER) -> User:
+        now = _now()
+        await self._collection.update_one(
+            {"mobile": mobile},
+            {
+                "$setOnInsert": {
+                    "mobile": mobile,
+                    "role": role.value,
+                    "is_active": True,
+                    "created_at": now,
+                },
+                "$set": {"updated_at": now},
+            },
+            upsert=True,
+        )
+        user = await self.get_by_mobile(mobile)
+        if user is None:
+            raise RuntimeError("failed to create user")
+        return user
+
+    async def ensure_admin_user(self, mobile: str) -> User:
+        now = _now()
+        await self._collection.update_one(
+            {"mobile": mobile},
+            {
+                "$set": {"role": Role.ADMIN.value, "is_active": True, "updated_at": now},
+                "$setOnInsert": {"mobile": mobile, "created_at": now},
+            },
+            upsert=True,
+        )
+        user = await self.get_by_mobile(mobile)
+        if user is None:
+            raise RuntimeError("failed to seed admin user")
+        return user
+
+
+class MongoRefreshSessionRepository(RefreshSessionRepository):
+    def __init__(self, db: AsyncIOMotorDatabase[Any]) -> None:
+        self._collection = db.refresh_sessions
+
+    async def create_indexes(self) -> None:
+        await self._collection.create_index([("token_hash", ASCENDING)], unique=True)
+        await self._collection.create_index([("user_id", ASCENDING), ("expires_at", ASCENDING)])
+        await self._collection.create_index([("revoked_at", ASCENDING)])
+
+    async def create(self, user_id: str, token_hash: str, expires_at: datetime) -> RefreshSession:
+        now = _now()
+        result = await self._collection.insert_one(
+            {
+                "user_id": user_id,
+                "token_hash": token_hash,
+                "expires_at": expires_at,
+                "revoked_at": None,
+                "replaced_by_hash": None,
+                "created_at": now,
+            }
+        )
+        doc = await self._collection.find_one({"_id": result.inserted_id})
+        if doc is None:
+            raise RuntimeError("failed to create refresh session")
+        return _session_from_doc(doc)
+
+    async def get_active_by_hash(self, token_hash: str, now: datetime) -> RefreshSession | None:
+        doc = await self._collection.find_one(
+            {"token_hash": token_hash, "revoked_at": None, "expires_at": {"$gt": now}}
+        )
+        return _session_from_doc(doc) if doc else None
+
+    async def revoke(
+        self, token_hash: str, now: datetime, replaced_by_hash: str | None = None
+    ) -> None:
+        update: dict[str, Any] = {"revoked_at": now}
+        if replaced_by_hash is not None:
+            update["replaced_by_hash"] = replaced_by_hash
+        await self._collection.update_one(
+            {"token_hash": token_hash, "revoked_at": None},
+            {"$set": update},
+        )
+

src/gapido_auth/infrastructure/rabbitmq.py

@@ -0,0 +1,53 @@
+import json
+from dataclasses import asdict
+from typing import Any
+
+import aio_pika
+from aio_pika.abc import AbstractChannel, AbstractQueue, AbstractRobustConnection, DeliveryMode
+
+from gapido_auth.domain.entities import SmsJob
+from gapido_auth.domain.ports import SmsPublisher
+
+SMS_EXCHANGE = "gapido.sms"
+SMS_QUEUE = "gapido.sms.otp"
+SMS_DLQ = "gapido.sms.otp.dlq"
+SMS_ROUTING_KEY = "otp"
+SMS_DLX = "gapido.sms.dlx"
+
+
+async def connect_robust(url: str) -> AbstractRobustConnection:
+    return await aio_pika.connect_robust(url)
+
+
+async def declare_sms_topology(channel: AbstractChannel) -> tuple[Any, AbstractQueue, Any]:
+    exchange = await channel.declare_exchange(
+        SMS_EXCHANGE, aio_pika.ExchangeType.DIRECT, durable=True
+    )
+    dlx = await channel.declare_exchange(SMS_DLX, aio_pika.ExchangeType.DIRECT, durable=True)
+    queue = await channel.declare_queue(
+        SMS_QUEUE,
+        durable=True,
+        arguments={"x-dead-letter-exchange": SMS_DLX, "x-dead-letter-routing-key": SMS_ROUTING_KEY},
+    )
+    dlq = await channel.declare_queue(SMS_DLQ, durable=True)
+    await queue.bind(exchange, routing_key=SMS_ROUTING_KEY)
+    await dlq.bind(dlx, routing_key=SMS_ROUTING_KEY)
+    return exchange, queue, dlx
+
+
+class RabbitMqSmsPublisher(SmsPublisher):
+    def __init__(self, channel: AbstractChannel) -> None:
+        self._channel = channel
+        self._exchange: Any | None = None
+
+    async def publish(self, job: SmsJob) -> None:
+        if self._exchange is None:
+            self._exchange, _, _ = await declare_sms_topology(self._channel)
+
+        body = json.dumps(asdict(job)).encode()
+        message = aio_pika.Message(
+            body=body,
+            content_type="application/json",
+            delivery_mode=DeliveryMode.PERSISTENT,
+        )
+        await self._exchange.publish(message, routing_key=SMS_ROUTING_KEY)

src/gapido_auth/infrastructure/redis_otp_store.py

@@ -0,0 +1,67 @@
+from typing import Any, cast
+
+from redis.asyncio import Redis
+
+from gapido_auth.domain.errors import OtpAttemptsExceeded, OtpExpired
+from gapido_auth.domain.ports import OtpStore
+
+
+class RedisOtpStore(OtpStore):
+    def __init__(self, redis: Redis) -> None:
+        self._redis = redis
+
+    async def allow_request(self, key: str, limit: int, window_seconds: int) -> bool:
+        count = await self._redis.incr(key)
+        if count == 1:
+            await self._redis.expire(key, window_seconds)
+        return int(count) <= limit
+
+    async def store_otp(
+        self,
+        mobile: str,
+        purpose: str,
+        otp_hash: str,
+        ttl_seconds: int,
+        max_attempts: int,
+    ) -> None:
+        key = self._otp_key(mobile, purpose)
+        attempts_key = self._attempts_key(mobile, purpose)
+        async with self._redis.pipeline(transaction=True) as pipe:
+            pipe.hset(key, mapping={"hash": otp_hash, "max_attempts": str(max_attempts)})
+            pipe.expire(key, ttl_seconds)
+            pipe.delete(attempts_key)
+            pipe.expire(attempts_key, ttl_seconds)
+            await pipe.execute()
+
+    async def verify_otp(self, mobile: str, purpose: str, candidate_hash: str) -> bool:
+        key = self._otp_key(mobile, purpose)
+        attempts_key = self._attempts_key(mobile, purpose)
+
+        stored_hash = await cast(Any, self._redis.hget(key, "hash"))
+        if stored_hash is None:
+            raise OtpExpired("OTP expired or was not requested")
+
+        max_attempts_raw = await cast(Any, self._redis.hget(key, "max_attempts"))
+        max_attempts = int(max_attempts_raw or 5)
+        attempts = int(await self._redis.incr(attempts_key))
+        ttl = await self._redis.ttl(key)
+        if ttl > 0:
+            await self._redis.expire(attempts_key, ttl)
+        if attempts > max_attempts:
+            await self._redis.delete(key)
+            raise OtpAttemptsExceeded("OTP attempts exceeded")
+
+        expected = stored_hash.decode() if isinstance(stored_hash, bytes) else str(stored_hash)
+        if expected != candidate_hash:
+            return False
+
+        await self._redis.delete(key, attempts_key)
+        return True
+
+    @staticmethod
+    def _otp_key(mobile: str, purpose: str) -> str:
+        return f"otp:{mobile}:{purpose}"
+
+    @staticmethod
+    def _attempts_key(mobile: str, purpose: str) -> str:
+        return f"otp-attempts:{mobile}:{purpose}"

src/gapido_auth/infrastructure/worker.py

@@ -0,0 +1,78 @@
+import asyncio
+import json
+import logging
+from typing import Any, cast
+
+import aio_pika
+from aio_pika.abc import AbstractIncomingMessage
+
+from gapido_auth.config import get_settings
+from gapido_auth.domain.entities import SmsJob
+from gapido_auth.infrastructure.kavenegar_client import KavenegarSmsClient
+from gapido_auth.infrastructure.rabbitmq import (
+    SMS_ROUTING_KEY,
+    connect_robust,
+    declare_sms_topology,
+)
+
+logger = logging.getLogger(__name__)
+MAX_RETRIES = 3
+
+
+async def handle_message(
+    message: AbstractIncomingMessage,
+    client: KavenegarSmsClient,
+    exchange: Any,
+    dlx: Any,
+) -> None:
+    async with message.process(ignore_processed=True, requeue=False):
+        payload = json.loads(message.body.decode())
+        job = SmsJob(**payload)
+        retry_count = int(cast(int | str, (message.headers or {}).get("x-retry-count", 0)))
+
+        try:
+            await client.send_otp(job.mobile, job.code, job.template)
+        except Exception:
+            if retry_count >= MAX_RETRIES:
+                logger.exception("SMS job failed permanently for mobile=%s", job.mobile)
+                await dlx.publish(
+                    aio_pika.Message(
+                        body=message.body,
+                        content_type="application/json",
+                        headers={"x-retry-count": retry_count},
+                        delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
+                    ),
+                    routing_key=SMS_ROUTING_KEY,
+                )
+                return
+
+            logger.warning(
+                "SMS job failed for mobile=%s retry=%s", job.mobile, retry_count + 1, exc_info=True
+            )
+            await exchange.publish(
+                aio_pika.Message(
+                    body=message.body,
+                    content_type="application/json",
+                    headers={"x-retry-count": retry_count + 1},
+                    delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
+                ),
+                routing_key=SMS_ROUTING_KEY,
+            )
+
+
+async def main() -> None:
+    logging.basicConfig(level=logging.INFO)
+    settings = get_settings()
+    connection = await connect_robust(settings.rabbitmq_url)
+    async with connection:
+        channel = await connection.channel()
+        await channel.set_qos(prefetch_count=20)
+        exchange, queue, dlx = await declare_sms_topology(channel)
+        client = KavenegarSmsClient(settings.kavenegar_api_key)
+        await queue.consume(lambda message: handle_message(message, client, exchange, dlx))
+        logger.info("SMS worker started")
+        await asyncio.Future()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())