fix: make it launchable (#62)

Author: emptybutton

deploy/dev/nats/add_streams.sh

@@ -1,3 +1,5 @@
 #!/bin/bash
 
-nats -s nats://nats:4222 stream add --config /mnt/streams/user.json
+for stream_config_file in `ls -lx /mnt/streams`; do
+    nats -s nats://nats:4222 stream add --config /mnt/streams/$stream_config_file
+done

pyproject.toml

@@ -117,6 +117,7 @@ ignore = [
 
 [tool.ruff.lint.per-file-ignores]
 "src/ttt/infrastructure/sqlalchemy/tables/__init__.py" = ["F401"]
+"src/ttt/infrastructure/taskiq/tasks/__init__.py" = ["F401"]
 "src/ttt/infrastructure/adapters/*" = ["ARG002"]
 "src/ttt/presentation/adapters/*" = ["ARG002"]
 "src/ttt/presentation/*" = ["RUF001"]

src/ttt/application/game/game/make_ai_move_in_game.py

@@ -26,8 +26,7 @@
 class MakeAiMoveInGame:
     map_: Map
     games: Games
-    game_views: GameViews
-    users: Users
+    views: GameViews
     uuids: UUIDs
     randoms: Randoms
     ai_gateway: GameAiGateway
@@ -77,4 +76,4 @@ async def __call__(self, user_id: int, game_id: UUID, ai_id: UUID) -> None:
                 await self.map_(tracking)
                 await self.transaction.commit()
 
-                await self.game_views.game_view(game)
+                await self.views.game_view(game)

src/ttt/application/game/game/ports/game_views.py

@@ -25,13 +25,6 @@ async def no_current_game_view(
         /,
     ) -> None: ...
 
-    @abstractmethod
-    async def no_game_with_id_view(
-        self,
-        game_id: UUID,
-        /,
-    ) -> None: ...
-
     @abstractmethod
     async def game_already_complteted_view(
         self,

src/ttt/infrastructure/adapters/transaction.py

@@ -32,9 +32,16 @@ async def __aexit__(
         error: BaseException | None,
         traceback: TracebackType | None,
     ) -> None:
-        transaction = not_none(self._session.get_transaction())
+        transaction = self._session.get_transaction()
+
+        if transaction is None:
+            return
+
         with reraise_serialization_error():
-            await transaction.__aexit__(error_type, error, traceback)
+            if error is None:
+                await transaction.commit()
+            else:
+                await transaction.rollback()
 
     async def commit(self) -> None:
         transaction = not_none(self._session.get_transaction())
@@ -49,7 +56,7 @@ class InPostgresNotSerializableTransaction(NotSerializableTransaction):
     async def __aenter__(self) -> Self:
         assert_(not self._session.in_transaction())
         await self._session.connection(
-            execution_options={"isolation_level": "READ COMMITED"},
+            execution_options={"isolation_level": "READ COMMITTED"},
         )
         return self
 
@@ -59,8 +66,15 @@ async def __aexit__(
         error: BaseException | None,
         traceback: TracebackType | None,
     ) -> None:
-        transaction = not_none(self._session.get_transaction())
-        await transaction.__aexit__(error_type, error, traceback)
+        transaction = self._session.get_transaction()
+
+        if transaction is None:
+            return
+
+        if error is None:
+            await transaction.commit()
+        else:
+            await transaction.rollback()
 
     async def commit(self) -> None:
         transaction = not_none(self._session.get_transaction())
@@ -83,5 +97,12 @@ async def __aexit__(
         error: BaseException | None,
         traceback: TracebackType | None,
     ) -> None:
-        transaction = not_none(self._session.get_transaction())
-        await transaction.__aexit__(error_type, error, traceback)
+        transaction = self._session.get_transaction()
+
+        if transaction is None:
+            return
+
+        if error is None:
+            await transaction.commit()
+        else:
+            await transaction.rollback()

src/ttt/infrastructure/adapters/user_locks.py

@@ -1,4 +1,5 @@
-from pydantic.dataclasses import dataclass
+from dataclasses import dataclass
+
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 

src/ttt/infrastructure/alembic/versions/2dcb2be9e277_remove_is_true_from_ix_users_has_.py

@@ -0,0 +1,44 @@
+"""
+remove `IS TRUE` from `ix_users_has_matchmaking_waiting`.
+
+Revision ID: 2dcb2be9e277
+Revises: 6642791b3bf8
+Create Date: 2025-09-24 13:46:33.322702
+
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+
+
+revision: str = "2dcb2be9e277"
+down_revision: str | None = "6642791b3bf8"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.drop_index(
+        op.f("ix_users_has_matchmaking_waiting"),
+        table_name="users",
+        postgresql_where="(has_matchmaking_waiting IS TRUE)",
+    )
+    op.create_index(
+        op.f("ix_users_has_matchmaking_waiting"),
+        "users",
+        ["has_matchmaking_waiting"],
+        unique=False,
+        postgresql_where="has_matchmaking_waiting",
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_users_has_matchmaking_waiting"), table_name="users")
+    op.create_index(
+        op.f("ix_users_has_matchmaking_waiting"),
+        "users",
+        ["has_matchmaking_waiting"],
+        unique=False,
+        postgresql_where="(has_matchmaking_waiting IS TRUE)",
+    )

src/ttt/infrastructure/sqlalchemy/tables/user.py

@@ -122,7 +122,7 @@ class TableUser(Base[User]):
         Index(
             "ix_users_has_matchmaking_waiting",
             has_matchmaking_waiting,
-            postgresql_where=(has_matchmaking_waiting.is_(True)),
+            postgresql_where="has_matchmaking_waiting",
         ),
     )
 

src/ttt/infrastructure/taskiq/broker.py

@@ -1,64 +1,174 @@
-from collections.abc import AsyncGenerator
-from typing import NewType, Protocol
+from asyncio import Queue, TaskGroup, gather
+from collections.abc import AsyncGenerator, Awaitable, Callable
+from dataclasses import dataclass, field
+from typing import Any, overload
 
 from nats.aio.msg import Msg as NatsMessage
 from nats.errors import TimeoutError as NatsTimeoutError
 from nats.js import JetStreamContext
 from taskiq import (
     AckableMessage,
     AsyncBroker,
+    AsyncTaskiqDecoratedTask,
     BrokerMessage,
 )
 
 
-class PullSubscribe(Protocol):
+@dataclass(frozen=True)
+class PullSubscribe:
+    _func: Callable[
+        [JetStreamContext, str], Awaitable[JetStreamContext.PullSubscription],
+    ]
+
     async def __call__(
-        self, js: JetStreamContext, subject: str, /,
-    ) ->  JetStreamContext.PullSubscription: ...
+        self, js: JetStreamContext, subject: str,
+    ) -> JetStreamContext.PullSubscription:
+        return await self._func(js, subject)
 
 
-class NatsBroker(AsyncBroker):
-    _consumer: JetStreamContext.PullSubscription
-    js: JetStreamContext
+@dataclass
+class NatsQueue:
+    _subject: str
+    _pull_subscribe: PullSubscribe
+    _pull_consume_batch: int
+    _pull_consume_timeout: float | None
 
-    def __init__(
-        self,
-        subject: str,
-        pull_subscribe: PullSubscribe,
-        pull_consume_batch: int = 1,
-        pull_consume_timeout: float | None = 5,
-    ) -> None:
-        super().__init__()
-        self._subject = subject
-        self._pull_subscribe = pull_subscribe
-        self._pull_consume_batch = pull_consume_batch
-        self._pull_consume_timeout = pull_consume_timeout
+    _js: JetStreamContext = field(init=False)
+    _pull_subscription: JetStreamContext.PullSubscription = field(init=False)
 
-    async def startup(self) -> None:
-        await super().startup()
-        self._consumer = await self._pull_subscribe(self.js, self._subject)
+    async def startup(self, js: JetStreamContext) -> None:
+        self._js = js
+        self._pull_subscription = await self._pull_subscribe(js, self._subject)
 
-    async def kick(self, message: BrokerMessage) -> None:
-        await self.js.publish(
+    async def push(self, message: BrokerMessage) -> None:
+        await self._js.publish(
             self._subject,
             payload=message.message,
             headers=message.labels,
         )
 
-    async def listen(self) -> AsyncGenerator[AckableMessage]:
+    async def pull_to(self, output: Queue[AckableMessage]) -> None:
+        nats_messages: list[NatsMessage]
+
         while True:
             try:
-                nats_messages: list[NatsMessage] = await self._consumer.fetch(
+                nats_messages = await self._pull_subscription.fetch(
                     batch=self._pull_consume_batch,
                     timeout=self._pull_consume_timeout,
                 )
-                for nats_message in nats_messages:
-                    yield AckableMessage(
-                        data=nats_message.data,
-                        ack=nats_message.ack,
-                    )
             except NatsTimeoutError:
                 continue
 
+            ackable_messages = (
+                AckableMessage(
+                    data=nats_message.data,
+                    ack=nats_message.ack,
+                )
+                for nats_message in nats_messages
+            )
+            await gather(*(
+                output.put(ackable_message)
+                for ackable_message in ackable_messages
+            ))
+
+
+class NatsBroker(AsyncBroker):
+    js: JetStreamContext
+    pulling_queue: Queue[AckableMessage]
+
+    def __init__(
+        self,
+        default_subject: str = "taskiq.>",
+        default_pull_subscribe: PullSubscribe | None = None,
+        default_pull_consume_batch: int = 1,
+        default_pull_consume_timeout: float | None = 5,
+    ) -> None:
+        super().__init__()
+        self._default_subject = default_subject
+        self._default_pull_subscribe = (
+            default_pull_subscribe
+            or (lambda js, sub: js.pull_subscribe(
+                sub,
+                "taskiq",
+                "TASKIQ",
+            ))
+        )
+        self._default_pull_consume_batch = default_pull_consume_batch
+        self._default_pull_consume_timeout = default_pull_consume_timeout
+        self._queue_by_task_name = dict[str, NatsQueue]()
+
+    @overload
+    def task[**PmT, RT](
+        self,
+        task_name: Callable[PmT, RT],
+        **labels: Any,  # noqa: ANN401
+    ) -> AsyncTaskiqDecoratedTask[PmT, RT]:
+        ...
+
+    @overload
+    def task[**PmT, RT](
+        self,
+        task_name: str | None = None,
+        **labels: Any,  # noqa: ANN401
+    ) -> Callable[
+        [Callable[PmT, RT]],
+        AsyncTaskiqDecoratedTask[PmT, RT],
+    ]:
+        ...
+
+    def task[**PmT, RT](
+        self,
+        task_name: str | Callable[PmT, RT] | None = None,
+        **labels: Any,
+    ) -> Any:
+        subject = labels.pop("subject", self._default_subject)
+        pull_subscribe = labels.pop(
+            "pull_subscribe", self._default_pull_subscribe,
+        )
+        pull_consume_batch = labels.pop(
+            "pull_consume_batch", self._default_pull_consume_batch,
+        )
+        pull_consume_timeout = labels.pop(
+            "pull_consume_timeout", self._default_pull_consume_timeout,
+        )
+        queue = NatsQueue(
+            subject,
+            pull_subscribe,
+            pull_consume_batch,
+            pull_consume_timeout,
+        )
+
+        result = super().task(task_name, **labels)
+
+        if isinstance(result, AsyncTaskiqDecoratedTask):
+            self._queue_by_task_name[result.task_name] = queue
+            return result
+
+        def decorator(
+            func: Callable[PmT, RT],
+        ) -> AsyncTaskiqDecoratedTask[PmT, RT]:
+            task = result(func)
+            self._queue_by_task_name[task.task_name] = queue
+            return task
+
+        return decorator
+
+    async def startup(self) -> None:
+        await super().startup()
+        await gather(*(
+            queue.startup(self.js)
+            for queue in self._queue_by_task_name.values()
+        ))
+
+    async def kick(self, message: BrokerMessage) -> None:
+        await self._queue_by_task_name[message.task_name].push(message)
+
+    async def listen(self) -> AsyncGenerator[AckableMessage]:
+        async with TaskGroup() as pulling_tasks:
+            for nats_queue in self._queue_by_task_name.values():
+                pulling_tasks.create_task(
+                    nats_queue.pull_to(self.pulling_queue),
+                )
 
-NatsBrokers = NewType("NatsBrokers", tuple[NatsBroker, ...])
+            while True:
+                yield await self.pulling_queue.get()

src/ttt/infrastructure/taskiq/tasks/__init__.py

@@ -0,0 +1,7 @@
+from ttt.infrastructure.taskiq.tasks.common import nats_tasks
+from ttt.infrastructure.taskiq.tasks.complete_stars_purchase_payment_task import (  # noqa: E501
+    complete_stars_purchase_payment_task,
+)
+from ttt.infrastructure.taskiq.tasks.make_ai_move_in_game_task import (
+    make_ai_move_in_game_broker_task,
+)